215 lines
7.8 KiB
Python
215 lines
7.8 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Focused semantic keyframes for DirectCodeEditor terminal presentation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import FrozenInstanceError
|
|
from multiprocessing.queues import Queue
|
|
from queue import Empty
|
|
|
|
import pytest
|
|
|
|
from band_aider.strategies.compiler import CompileError, TypeCheckError
|
|
from band_aider.judgement.direct_code_editor import _diagnostic_movement
|
|
from band_aider.ui.dto.direct_code_editor_status import DirectCodeEditorStatusDTO
|
|
from band_aider.ui.dto.log_message import LogMessageDTO
|
|
from band_aider.ui.dto.supreme_status import SupremeStatusDTO
|
|
from band_aider.ui.terminal_ui import (
|
|
TerminalUI,
|
|
render_direct_code_editor_lines,
|
|
render_projection,
|
|
)
|
|
|
|
|
|
class _MemoryQueue(Queue[object]):
|
|
"""Provide the terminal's nominal queue contract without worker resources."""
|
|
|
|
def __init__(self, events: list[object] | None = None) -> None:
|
|
"""Store stationary event and command values in process-local memory."""
|
|
|
|
self._events = list(events or [])
|
|
self.entries: list[object] = []
|
|
|
|
def get_nowait(self) -> object:
|
|
"""Return one queued event or match the nonblocking queue contract."""
|
|
|
|
if not self._events:
|
|
raise Empty
|
|
return self._events.pop(0)
|
|
|
|
def put(
|
|
self,
|
|
obj: object,
|
|
block: bool = True,
|
|
timeout: float | None = None,
|
|
) -> None:
|
|
"""Record a terminal command without involving a worker."""
|
|
|
|
del block, timeout
|
|
self.entries.append(obj)
|
|
|
|
|
|
def _status(**changes: object) -> DirectCodeEditorStatusDTO:
|
|
"""Create a complete presentation snapshot with focused override values."""
|
|
|
|
data = {
|
|
"editable_path": "tests/test_widget.py",
|
|
"repair_turn": 18,
|
|
"phase": "candidate compile",
|
|
"raw_compiler_error_count": 31,
|
|
"blocking_compiler_error_count": 23,
|
|
"model_visible_compiler_error_count": 1,
|
|
"static_type_check_error_count": 20,
|
|
"unresolved_targeted_test_failure_count": 0,
|
|
}
|
|
data.update(changes)
|
|
return DirectCodeEditorStatusDTO(**data)
|
|
|
|
|
|
def _active_event(**changes: object) -> SupremeStatusDTO:
|
|
"""Wrap one DirectCodeEditor snapshot in its established status event."""
|
|
|
|
return SupremeStatusDTO(direct_code_editor=_status(**changes))
|
|
|
|
|
|
def test_direct_code_editor_status_is_immutable() -> None:
|
|
"""A published progress snapshot cannot be changed after it enters the UI queue."""
|
|
|
|
status = _status(named_targeted_failures=("FAIL TestWidget.test_render",))
|
|
|
|
with pytest.raises(FrozenInstanceError):
|
|
setattr(status, "phase", "decision")
|
|
|
|
|
|
def test_active_editor_defaults_to_progress_and_f7_only_changes_presentation() -> None:
|
|
"""Activation hides logs, then F7 round-trips the existing log focus without worker commands."""
|
|
|
|
commands = _MemoryQueue()
|
|
ui = TerminalUI(_MemoryQueue([_active_event()]), commands)
|
|
ui._set_focus("filter")
|
|
ui._drain_events()
|
|
|
|
assert ui._direct_editor_progress_visible() is True
|
|
assert ui._focused_field() == "editor"
|
|
assert "filter" not in ui._visible_focus_fields()
|
|
|
|
ui._toggle_direct_editor_view()
|
|
|
|
assert ui._direct_editor_progress_visible() is False
|
|
assert ui._focused_field() == "filter"
|
|
assert commands.entries == []
|
|
|
|
ui._toggle_direct_editor_view()
|
|
|
|
assert ui._direct_editor_progress_visible() is True
|
|
assert ui._focused_field() == "editor"
|
|
assert commands.entries == []
|
|
|
|
|
|
def test_new_editor_activation_resets_to_progress_and_hidden_logs_accumulate() -> None:
|
|
"""A later editor episode begins on progress while the ordinary log ring keeps every event."""
|
|
|
|
ui = TerminalUI(_MemoryQueue([_active_event()]), _MemoryQueue())
|
|
ui._set_focus("logs")
|
|
ui._drain_events()
|
|
ui.projection.apply(LogMessageDTO(1, "INFO", "editor", "first retained log"))
|
|
ui._toggle_direct_editor_view()
|
|
ui._sync_direct_editor_mode(SupremeStatusDTO(direct_code_editor=None))
|
|
assert ui._focused_field() == "logs"
|
|
ui._sync_direct_editor_mode(_active_event())
|
|
ui.projection.apply(LogMessageDTO(2, "INFO", "editor", "second retained log"))
|
|
|
|
assert ui._direct_editor_progress_visible() is True
|
|
assert [item.message for item in ui.projection.logs] == ["first retained log", "second retained log"]
|
|
|
|
|
|
def test_progress_panel_replaces_log_focus_until_f7_restores_it() -> None:
|
|
"""The active editor keeps log controls hidden until its presentation toggle."""
|
|
|
|
ui = TerminalUI(_MemoryQueue([_active_event()]), _MemoryQueue())
|
|
ui._drain_events()
|
|
assert ui._direct_editor_progress_visible() is True
|
|
assert "logs" not in ui._visible_focus_fields()
|
|
|
|
ui._toggle_direct_editor_view()
|
|
|
|
assert ui._direct_editor_progress_visible() is False
|
|
assert "logs" in ui._visible_focus_fields()
|
|
|
|
|
|
def test_editor_message_does_not_change_active_progress_state() -> None:
|
|
"""An ephemeral message leaves the active editor presentation available."""
|
|
|
|
ui = TerminalUI(_MemoryQueue([_active_event()]), _MemoryQueue())
|
|
ui._set_focus("logs")
|
|
ui._drain_events()
|
|
ui._message = "long ephemeral message " * 10
|
|
|
|
assert ui._direct_editor_progress_visible() is True
|
|
assert ui._focused_field() == "editor"
|
|
|
|
|
|
def test_f3_and_f4_cannot_focus_hidden_log_controls() -> None:
|
|
"""Direct progress keeps filter and logger controls unavailable until logs are visible."""
|
|
|
|
ui = TerminalUI(_MemoryQueue([_active_event()]), _MemoryQueue())
|
|
ui._set_focus("logs")
|
|
ui._drain_events()
|
|
|
|
ui._set_focus("filter")
|
|
assert ui._focused_field() == "editor"
|
|
assert ui._message == "selected editor"
|
|
ui._set_focus("logger")
|
|
assert ui._focused_field() == "editor"
|
|
|
|
|
|
def test_progress_lines_label_counts_and_targeted_results_without_inference() -> None:
|
|
"""The panel distinguishes error counts and shows only compiler-provided target identities."""
|
|
|
|
lines = render_direct_code_editor_lines(
|
|
_status(
|
|
unresolved_targeted_test_failure_count=3,
|
|
named_targeted_failures=("FAIL TestWidget.test_render", "FAIL TestWidget.test_save"),
|
|
unattributed_targeted_execution_failure_count=1,
|
|
patch_outcome="partial",
|
|
unresolved_patch_issue_count=1,
|
|
)
|
|
)
|
|
|
|
assert any("23 blocking remaining · 31 raw · 1 shown to model" in line for line in lines)
|
|
assert "FAIL TestWidget.test_render" in lines
|
|
assert "FAIL TestWidget.test_save" in lines
|
|
assert "UNKNOWN targeted result not established" in lines
|
|
assert not any("test_unknown" in line for line in lines)
|
|
|
|
|
|
def test_diagnostic_movement_distinguishes_identity_and_coordinates() -> None:
|
|
"""Observability identifies substitutions even when equal diagnostic counts leave policy unchanged."""
|
|
|
|
original = [TypeCheckError("missing A", path=None, line=2, column=3)]
|
|
|
|
assert _diagnostic_movement(None, original) == "initial"
|
|
assert _diagnostic_movement(original, original) == "same diagnostic set"
|
|
assert _diagnostic_movement(original, [TypeCheckError("missing A", line=4, column=1)]) == "coordinates changed"
|
|
assert _diagnostic_movement(original, [CompileError("missing B")]) == "diagnostics changed"
|
|
assert _diagnostic_movement(original, []) == "clean"
|
|
|
|
|
|
def test_plain_rendering_replaces_log_excerpt_only_while_editor_is_active() -> None:
|
|
"""Plain output presents progress during an active editor snapshot and keeps inactive logs unchanged."""
|
|
|
|
ui = TerminalUI(_MemoryQueue(), _MemoryQueue())
|
|
ui.projection.apply(LogMessageDTO(1, "INFO", "editor", "retained log"))
|
|
ui.projection.status = _active_event()
|
|
|
|
active = render_projection(ui.projection)
|
|
ui.projection.status = SupremeStatusDTO()
|
|
inactive = render_projection(ui.projection)
|
|
|
|
assert "DIRECT CODE EDITOR" in active
|
|
assert "retained log" not in active
|
|
assert "LOGS" in inactive
|
|
assert "retained log" in inactive
|