mirror of
https://github.com/trustedsec/hate_crack.git
synced 2026-07-28 14:47:22 -07:00
Justin BollingerandClaude
e5e1107359
Defect 1 — hcatOllama showed a plain print then blocked silently for up to 300 s while waiting for the LLM. A new generic context manager `hate_crack/progress.py::spinner()` now runs a daemon thread that repaints a single line with a frame + elapsed-seconds counter every ~120 ms. TTY guard ensures non-TTY/test environments get a single plain print instead of ANSI control characters. The line is erased with \033[2K\r on exit (normal and exceptional). Defect 2 — the wordlist path materialised every line in memory and built a prompt that could massively overflow ollamaNumCtx on large wordlists. The path now does a two-pass evenly-spaced sample: pass 1 counts usable lines, pass 2 stride-selects up to `ollamaMaxSampleLines` (default 500, new config key) spread across the full file. When no capping occurs the message reads "Loaded N passwords"; when capping occurs it reads "Sampled N of M passwords". Co-Authored-By: Claude <noreply@anthropic.com>
109 lines
3.8 KiB
Python
109 lines
3.8 KiB
Python
"""Tests for hate_crack/progress.py — the generic spinner context manager."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import sys
|
|
import threading
|
|
import time
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
from hate_crack.progress import spinner
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TTY guard: non-TTY path must print message once and not start a thread
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_non_tty_prints_message(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""In a non-TTY environment the message is printed once and no thread runs."""
|
|
with mock.patch.object(sys, "stdout", wraps=sys.stdout) as m_stdout:
|
|
m_stdout.isatty.return_value = False # type: ignore[attr-defined]
|
|
# Use actual capsys for the print capture but override isatty
|
|
original_isatty = sys.stdout.isatty
|
|
sys.stdout.isatty = lambda: False # type: ignore[method-assign]
|
|
try:
|
|
with spinner("Working..."):
|
|
pass
|
|
finally:
|
|
sys.stdout.isatty = original_isatty # type: ignore[method-assign]
|
|
|
|
captured = capsys.readouterr()
|
|
assert "Working..." in captured.out
|
|
|
|
|
|
def test_non_tty_no_extra_thread() -> None:
|
|
"""Spinner in non-TTY context should not start a background thread."""
|
|
buf = io.StringIO()
|
|
buf.isatty = lambda: False # type: ignore[attr-defined]
|
|
|
|
threads_before = threading.active_count()
|
|
with mock.patch("sys.stdout", buf):
|
|
with spinner("No threads please"):
|
|
peak = threading.active_count()
|
|
# The thread count during the body should not exceed baseline + 1
|
|
# (pytest itself may add threads; we just want no extra daemon spinner thread).
|
|
assert peak <= threads_before + 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TTY path: spinner starts a thread, clears the line on exit
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_tty_clears_line_on_exit() -> None:
|
|
"""After the context exits the spinner line must be erased."""
|
|
buf = io.StringIO()
|
|
buf.isatty = lambda: True # type: ignore[attr-defined]
|
|
|
|
with mock.patch("sys.stdout", buf):
|
|
with spinner("Testing..."):
|
|
time.sleep(0.05) # let spinner tick at least once
|
|
|
|
output = buf.getvalue()
|
|
# Line clear sequence must be present somewhere after the spinner started.
|
|
assert "\033[2K\r" in output or ("\033[2K" in output)
|
|
|
|
|
|
def test_tty_shows_elapsed_seconds() -> None:
|
|
"""The spinner should show at least a 0s counter in its output."""
|
|
buf = io.StringIO()
|
|
buf.isatty = lambda: True # type: ignore[attr-defined]
|
|
|
|
with mock.patch("sys.stdout", buf):
|
|
with spinner("Counting..."):
|
|
time.sleep(0.25) # enough time for several ticks
|
|
|
|
output = buf.getvalue()
|
|
# Should have written at least one elapsed counter like "0s" or "1s"
|
|
assert "s" in output
|
|
assert "Counting..." in output
|
|
|
|
|
|
def test_tty_exception_still_clears_line() -> None:
|
|
"""Spinner must clear the line even when the body raises."""
|
|
buf = io.StringIO()
|
|
buf.isatty = lambda: True # type: ignore[attr-defined]
|
|
|
|
with mock.patch("sys.stdout", buf):
|
|
with pytest.raises(ValueError):
|
|
with spinner("Will raise"):
|
|
raise ValueError("boom")
|
|
|
|
output = buf.getvalue()
|
|
assert "\033[2K\r" in output or ("\033[2K" in output)
|
|
|
|
|
|
def test_spinner_does_not_swallow_exception() -> None:
|
|
"""The spinner context manager must re-raise exceptions from the body."""
|
|
buf = io.StringIO()
|
|
buf.isatty = lambda: True # type: ignore[attr-defined]
|
|
|
|
with mock.patch("sys.stdout", buf):
|
|
with pytest.raises(RuntimeError, match="test error"):
|
|
with spinner("Should propagate"):
|
|
raise RuntimeError("test error")
|