51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""In-memory worker-stream capture contracts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from multiprocessing.queues import Queue
|
|
|
|
from band_aider.ui.dto.log_message import LogMessageDTO
|
|
from band_aider.ui.stream_capture import QueueTextWriter
|
|
|
|
|
|
class _MemoryQueue(Queue[object]):
|
|
"""Use Queue's nominal boundary while retaining messages only in local memory."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Avoid constructing the operating-system queue used by worker processes."""
|
|
|
|
self.entries: list[object] = []
|
|
|
|
def put(
|
|
self,
|
|
obj: object,
|
|
block: bool = True,
|
|
timeout: float | None = None,
|
|
) -> None:
|
|
"""Record a worker message without process or filesystem communication."""
|
|
|
|
del block, timeout
|
|
self.entries.append(obj)
|
|
|
|
|
|
def test_queue_text_writer_emits_only_complete_nonblank_lines_from_memory() -> None:
|
|
"""Partial output remains buffered until flush while blank lines never become log events."""
|
|
|
|
queue = _MemoryQueue()
|
|
writer = QueueTextWriter(queue, logger_name="stationary.worker", level="DEBUG")
|
|
|
|
assert writer.write("") == 0
|
|
assert writer.write("first line\n\nsecond line") == len("first line\n\nsecond line")
|
|
writer.flush()
|
|
writer.flush()
|
|
|
|
entries = [entry for entry in queue.entries if isinstance(entry, LogMessageDTO)]
|
|
assert writer.isatty() is False
|
|
assert [(entry.level, entry.logger_name, entry.message) for entry in entries] == [
|
|
("DEBUG", "stationary.worker", "first line"),
|
|
("DEBUG", "stationary.worker", "second line"),
|
|
]
|