fix: smooth GUI during pipeline downloads and harden wizard connection test
GUI log batching (_poll_log now drains queue into a single CTkTextbox.insert
call per 80 ms tick instead of N calls, each with see("end") scroll).
_QueueWriter strips ANSI/CSI escape codes and bare \r before enqueuing so
tqdm progress output is legible in the log textbox. OSC sequences terminated
by both BEL (\x07) and ST (\x1b\) are handled.
Wizard "Test Connection" moved off the main thread: requests.get runs in a
daemon thread; result posted back via after(0, ...). Widget refs captured
before thread launch to prevent stale updates if user navigates away. Bare
except narrowed to TclError (destroyed-widget guard only).
Code quality: import os moved to module level in app.py; _read_raw_config()
helper extracted to deduplicate dual raw config.json reads; return type
annotations added to _get_view_class, _get_dashboard, and cfg property.
Tests: 11 new unit tests for _QueueWriter (RED -> GREEN on OSC-ST fix).
This commit is contained in:
111
test_suite.py
111
test_suite.py
@@ -1353,6 +1353,117 @@ test("end-to-end offline pipeline (parse → compare → report)", _test_end_to_
|
||||
test("comparison.json on disk matches fresh parse+compare", _test_comparison_json_consistent_with_html)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 11. gui._io — QueueWriter
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
group("gui._io — QueueWriter")
|
||||
|
||||
import importlib.util as _ilu
|
||||
import queue as _queue_mod
|
||||
|
||||
# Load gui/_io.py without triggering gui/__init__.py (which requires customtkinter)
|
||||
_io_spec = _ilu.spec_from_file_location("gui._io", Path(__file__).parent / "gui" / "_io.py")
|
||||
_io_mod = _ilu.module_from_spec(_io_spec)
|
||||
_io_spec.loader.exec_module(_io_mod)
|
||||
_QueueWriter = _io_mod._QueueWriter
|
||||
|
||||
|
||||
def _qw():
|
||||
"""Return a fresh (writer, queue) pair."""
|
||||
q = _queue_mod.Queue()
|
||||
return _QueueWriter(q), q
|
||||
|
||||
|
||||
def _test_qw_clean_text_passes_through():
|
||||
w, q = _qw()
|
||||
w.write("hello\n")
|
||||
assert_eq(q.get_nowait(), "hello\n")
|
||||
|
||||
|
||||
def _test_qw_strips_csi_escape_sequences():
|
||||
w, q = _qw()
|
||||
w.write("\x1b[32mGreen\x1b[0m")
|
||||
assert_eq(q.get_nowait(), "Green")
|
||||
|
||||
|
||||
def _test_qw_strips_osc_escape_sequences():
|
||||
w, q = _qw()
|
||||
w.write("\x1b]0;window title\x07visible text")
|
||||
assert_eq(q.get_nowait(), "visible text")
|
||||
|
||||
|
||||
def _test_qw_bare_cr_becomes_newline():
|
||||
w, q = _qw()
|
||||
w.write("first\rsecond")
|
||||
assert_eq(q.get_nowait(), "first\nsecond")
|
||||
|
||||
|
||||
def _test_qw_crlf_normalised_to_lf():
|
||||
w, q = _qw()
|
||||
w.write("line1\r\nline2")
|
||||
assert_eq(q.get_nowait(), "line1\nline2")
|
||||
|
||||
|
||||
def _test_qw_empty_string_not_enqueued():
|
||||
w, q = _qw()
|
||||
w.write("")
|
||||
assert q.empty(), "empty write should not enqueue anything"
|
||||
|
||||
|
||||
def _test_qw_only_ansi_not_enqueued():
|
||||
w, q = _qw()
|
||||
w.write("\x1b[32m\x1b[0m")
|
||||
assert q.empty(), "write that strips to empty should not enqueue"
|
||||
|
||||
|
||||
def _test_qw_returns_original_byte_length():
|
||||
w, q = _qw()
|
||||
raw = "\x1b[32mhello\x1b[0m"
|
||||
result = w.write(raw)
|
||||
assert_eq(result, len(raw))
|
||||
|
||||
|
||||
def _test_qw_tqdm_progress_line():
|
||||
"""tqdm uses \\r to overwrite in-place; should become a newline in the log."""
|
||||
w, q = _qw()
|
||||
w.write("\r 50%|█████ | 1/2 [00:01<00:01]")
|
||||
out = q.get_nowait()
|
||||
assert out.startswith("\n"), f"expected leading newline, got {out!r}"
|
||||
assert "50%" in out
|
||||
|
||||
|
||||
def _test_qw_mixed_ansi_and_cr():
|
||||
w, q = _qw()
|
||||
w.write("\x1b[1m\rProgress: 75%\x1b[0m")
|
||||
out = q.get_nowait()
|
||||
assert "\x1b" not in out, "ANSI codes should be stripped"
|
||||
assert "\r" not in out, "bare CR should be converted"
|
||||
assert "75%" in out
|
||||
|
||||
|
||||
test("clean text passes through unchanged", _test_qw_clean_text_passes_through)
|
||||
test("CSI escape sequences stripped", _test_qw_strips_csi_escape_sequences)
|
||||
test("OSC escape sequences stripped", _test_qw_strips_osc_escape_sequences)
|
||||
test("bare CR converted to newline", _test_qw_bare_cr_becomes_newline)
|
||||
test("CRLF normalised to LF", _test_qw_crlf_normalised_to_lf)
|
||||
test("empty string not enqueued", _test_qw_empty_string_not_enqueued)
|
||||
test("all-ANSI write not enqueued", _test_qw_only_ansi_not_enqueued)
|
||||
test("write() returns original length", _test_qw_returns_original_byte_length)
|
||||
test("tqdm \\r progress line becomes newline", _test_qw_tqdm_progress_line)
|
||||
test("mixed ANSI + CR stripped and converted", _test_qw_mixed_ansi_and_cr)
|
||||
|
||||
|
||||
def _test_qw_osc_st_terminator():
|
||||
"""OSC sequences ended with ST (ESC \\) must also be stripped."""
|
||||
w, q = _qw()
|
||||
w.write("\x1b]0;title\x1b\\visible text")
|
||||
assert_eq(q.get_nowait(), "visible text")
|
||||
|
||||
|
||||
test("OSC ST-terminated sequence stripped", _test_qw_osc_st_terminator)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Summary
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user