mirror of
https://github.com/trustedsec/hate_crack.git
synced 2026-07-28 14:47:22 -07:00
Merge branch 'main' into feature/llm-atomic-agents
Conflicts resolved: - CHANGELOG.md: kept both [2.12.0] (branch) and [2.11.4] (main), newest first - pyproject.toml: kept main's tighter version pins (click>=8.4.2, requests>=2.34.2, packaging>=26.2, pytest-cov==7.1.0) and the branch's atomic-agents>=2.0.0 dep Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -190,6 +190,49 @@ class TestCrackTailerStop:
|
||||
tailer.stop()
|
||||
|
||||
|
||||
class TestCrackTailerPollOnceInternals:
|
||||
"""Direct unit tests for _poll_once edge cases via the internal API."""
|
||||
|
||||
def _make_uninitialised(self, out_path: Path):
|
||||
"""Return a tailer whose _file_pos is still None (seek never called)."""
|
||||
tailer, notify, aggregate = _make_tailer(out_path)
|
||||
# Do NOT call start() or _seek_to_eof(); leave _file_pos = None.
|
||||
return tailer, notify, aggregate
|
||||
|
||||
def test_first_poll_with_file_pos_none_seeds_position(
|
||||
self, tmp_path: Path
|
||||
) -> None:
|
||||
"""_poll_once with _file_pos=None must record size and return without notifying."""
|
||||
out = tmp_path / "hashes.out"
|
||||
out.write_text("alice:hash:plain\n")
|
||||
tailer, notify, aggregate = self._make_uninitialised(out)
|
||||
assert tailer._file_pos is None
|
||||
tailer._poll_once()
|
||||
# Position seeded to current size; no notification fired.
|
||||
assert tailer._file_pos == out.stat().st_size
|
||||
assert notify.call_count == 0
|
||||
assert aggregate.call_count == 0
|
||||
|
||||
def test_truncation_resets_position_directly(self, tmp_path: Path) -> None:
|
||||
"""_poll_once must reset position and buffer when the file shrinks."""
|
||||
out = tmp_path / "hashes.out"
|
||||
content = "alice:hash:plain\nbob:hash:plain\n"
|
||||
out.write_text(content)
|
||||
tailer, notify, aggregate = _make_tailer(out)
|
||||
tailer._file_pos = len(content.encode()) # pretend we've read it all
|
||||
tailer._buffer = b"leftover"
|
||||
|
||||
# Truncate to a smaller file.
|
||||
out.write_text("c:h:p\n")
|
||||
tailer._poll_once()
|
||||
|
||||
# Position must have been reset and buffer cleared.
|
||||
assert tailer._file_pos == len(b"c:h:p\n")
|
||||
assert tailer._buffer == b""
|
||||
# One new line notified.
|
||||
assert notify.call_count == 1
|
||||
|
||||
|
||||
class TestCrackTailerFileHandling:
|
||||
def test_missing_file_then_appearing(self, tmp_path: Path) -> None:
|
||||
out = tmp_path / "missing.out"
|
||||
|
||||
Reference in New Issue
Block a user