From d487996281f71ccd3034dd708fa099057e528b42 Mon Sep 17 00:00:00 2001 From: historia Date: Fri, 28 Aug 2026 03:16:14 -0400 Subject: feat: unified logging with logging_kit.py --- app/tests/test_hub.py | 7 ++- app/tests/test_logging_kit.py | 118 +++++++++++++++++++++++++++++++++++ app/tests/test_runview.py | 65 ++++++++++++++++++++ app/tests/test_taskview.py | 140 +++++++++++++++++++++++++++++++++++++++++- 4 files changed, 324 insertions(+), 6 deletions(-) create mode 100644 app/tests/test_logging_kit.py (limited to 'app/tests') diff --git a/app/tests/test_hub.py b/app/tests/test_hub.py index f0af74c..0bdf2d5 100644 --- a/app/tests/test_hub.py +++ b/app/tests/test_hub.py @@ -2535,10 +2535,11 @@ class SettingsTests(unittest.TestCase): # The ports section note hangs off the first port field, the remote # section note off the first remote URL field. notes = {f["key"]: f.get("note") for f in captured["fields"]} - self.assertTrue(notes["input_dir"]) - self.assertTrue(notes["output_dir"]) - self.assertTrue(notes["debug"]) + self.assertIsNone(notes["input_dir"]) + self.assertIsNone(notes["output_dir"]) + self.assertIsNone(notes["debug"]) self.assertTrue(notes["stop_and_exit"]) + self.assertTrue(notes["unload_models"]) self.assertTrue(notes["audiocpp_port"]) self.assertTrue(notes["audiocpp_remote_url"]) self.assertIsNone(notes["audio_format"]) diff --git a/app/tests/test_logging_kit.py b/app/tests/test_logging_kit.py new file mode 100644 index 0000000..1945e5e --- /dev/null +++ b/app/tests/test_logging_kit.py @@ -0,0 +1,118 @@ +"""Tests for app/logging_kit.py — log naming, teeing, and retention.""" + +import io +import os +import tempfile +import time +import unittest +from pathlib import Path + +import logging_kit + + +class _TempDir(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.dir = Path(self.tmp.name) + + +class DayStreamTests(_TempDir): + def test_creates_dated_stream_and_appends(self): + handle = logging_kit.day_stream("tui", log_dir=self.dir) + self.assertIsNotNone(handle) + logging_kit.write_line(handle, "first") + logging_kit.write_line(handle, "second") + handle.close() + files = list(self.dir.glob("tui_*.log")) + self.assertEqual(len(files), 1) + self.assertRegex(files[0].name, r"^tui_\d{8}\.log$") + self.assertEqual(files[0].read_text(encoding="utf-8"), + "first\nsecond\n") + + def test_unwritable_dir_returns_none(self): + blocker = self.dir / "file" # a file where the directory would be + blocker.write_text("x", encoding="utf-8") + self.assertIsNone( + logging_kit.day_stream("tui", log_dir=blocker)) + + +class RunArtifactTests(_TempDir): + def test_creates_timestamped_artifact(self): + path, handle = logging_kit.run_artifact("audiocpp_build", + log_dir=self.dir) + self.assertIsNotNone(handle) + self.assertRegex(path.name, r"^audiocpp_build_\d{8}_\d{6}\.log$") + logging_kit.write_line(handle, "out") + handle.close() + self.assertEqual(path.read_text(encoding="utf-8"), "out\n") + + def test_returns_path_even_when_unwritable(self): + blocker = self.dir / "file" + blocker.write_text("x", encoding="utf-8") + path, handle = logging_kit.run_artifact("x", log_dir=blocker) + self.assertIsNone(handle) + self.assertEqual(path.parent, blocker) + + +class WriteLineTests(unittest.TestCase): + def test_none_handle_is_a_noop(self): + logging_kit.write_line(None, "x") # must not raise + + def test_broken_handle_is_swallowed(self): + class Broken: + def write(self, _): + raise OSError("nope") + + def flush(self): + raise OSError("nope") + + logging_kit.write_line(Broken(), "x") # must not raise + + +class TeeWriterTests(unittest.TestCase): + def test_mirrors_to_both_sides(self): + logf, inner = io.StringIO(), io.StringIO() + writer = logging_kit.TeeWriter(logf, inner) + self.assertEqual(writer.write("hello"), 5) + writer.flush() + self.assertEqual(logf.getvalue(), "hello") + self.assertEqual(inner.getvalue(), "hello") + + def test_empty_write_returns_zero(self): + self.assertEqual(logging_kit.TeeWriter(io.StringIO()).write(""), 0) + + def test_none_sides_and_errors_are_swallowed(self): + writer = logging_kit.TeeWriter(None, None) + self.assertEqual(writer.write("x"), 1) # must not raise + writer.flush() + self.assertFalse(writer.isatty()) + + +class PruneTests(_TempDir): + def test_deletes_old_logs_keeps_new_server_logs_and_pid_files(self): + old = self.dir / "tui_20200101.log" + old.write_text("x", encoding="utf-8") + new = self.dir / "audiobook_20990101.log" + new.write_text("x", encoding="utf-8") + server = self.dir / "audiocpp-server.log" + server.write_text("x", encoding="utf-8") + pid = self.dir / "audiocpp-server.pid" + pid.write_text("1", encoding="utf-8") + ancient = time.time() - 100 * 86400 + os.utime(old, (ancient, ancient)) + os.utime(server, (ancient, ancient)) + + logging_kit.prune_logs(log_dir=self.dir) + + self.assertFalse(old.exists()) + self.assertTrue(new.exists()) + self.assertTrue(server.exists()) # owned by a server process + self.assertTrue(pid.exists()) # not a log + + def test_missing_dir_is_a_noop(self): + logging_kit.prune_logs(log_dir=self.dir / "nope") # must not raise + + +if __name__ == "__main__": + unittest.main() diff --git a/app/tests/test_runview.py b/app/tests/test_runview.py index f2d5180..8feeee1 100644 --- a/app/tests/test_runview.py +++ b/app/tests/test_runview.py @@ -10,6 +10,7 @@ cancel → stop-server flow. import os import sys import tempfile +import types import unittest from queue import Empty from unittest.mock import patch @@ -129,6 +130,70 @@ class StateTransitionTests(_FakeTui, unittest.TestCase): self.assertEqual(view.server, "stopped") +class LogAppenderTests(_FakeTui, unittest.TestCase): + """_LogAppender: stray console output survives in the run's log file.""" + + def _appender(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + path = os.path.join(tmp.name, "audiobook_20260828.log") + return path, runview._LogAppender(path) + + def test_write_splits_lines_into_the_file(self): + path, appender = self._appender() + appender.write("one\ntwo\n") + appender.write("three") + appender.flush() + with open(path, encoding="utf-8") as handle: + lines = handle.read().splitlines() + self.assertEqual(len(lines), 3) + self.assertTrue(lines[0].endswith(" - one")) + self.assertTrue(lines[2].endswith(" - three")) + + def test_blank_lines_and_empty_path_are_skipped(self): + path, appender = self._appender() + appender.write("\n\n") + appender.flush() + appender.write("x") + appender.flush() + empty = runview._LogAppender("") + empty.write("ignored\n") # must not raise + with open(path, encoding="utf-8") as handle: + self.assertEqual(len(handle.read().splitlines()), 1) + + def test_unwritable_path_never_raises(self): + appender = runview._LogAppender("/nonexistent-dir-zz/log.log") + appender.write("boom\n") # must not raise + appender.flush() + + def test_worker_stdout_is_mirrored_to_the_run_log(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + log_path = os.path.join(tmp.name, "audiobook_20260828.log") + view, _ = self.make_view(log_path=log_path) + + fake = types.ModuleType("audiobook") + + def convert(**_kwargs): + print("[INFO] stray console output") + + fake.convert = convert + with patch.dict(sys.modules, {"audiobook": fake}): + view._worker_main() + + with open(log_path, encoding="utf-8") as handle: + text = handle.read() + self.assertIn("[INFO] stray console output", text) + # The worker signed off normally through the event queue. + kinds = [] + while True: + try: + kinds.append(view._queue.get_nowait()["kind"]) + except Empty: + break + self.assertEqual(kinds, ["worker_exit"]) + + class RenderTests(_FakeTui, unittest.TestCase): def _strings(self, screen): return " ".join(text for _, _, text, _ in screen.strings) diff --git a/app/tests/test_taskview.py b/app/tests/test_taskview.py index 646fccb..1db708d 100644 --- a/app/tests/test_taskview.py +++ b/app/tests/test_taskview.py @@ -9,8 +9,10 @@ transitions through ``handle_event`` + ``_step_mark`` + ``_result_rc``. import io import sys +import tempfile import threading import unittest +from pathlib import Path from queue import Empty from unittest.mock import patch @@ -30,6 +32,12 @@ class _FakeTui: patcher = patch.dict(sys.modules, {"curses": self.curses}) patcher.start() self.addCleanup(patcher.stop) + # Console mirroring is disabled by default so tests stay hermetic; + # the console-log tests patch day_stream with a real temp file. + log_patcher = patch.object(taskview.logging_kit, "day_stream", + return_value=None) + log_patcher.start() + self.addCleanup(log_patcher.stop) taskview.tui._THEME.clear() self.addCleanup(taskview.tui._THEME.clear) @@ -41,6 +49,18 @@ class _FakeTui: return view, screen +class _TempLog: + """A temp target for the tui_ day stream (patched over day_stream).""" + + def _log_target(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + path = Path(tmp.name) / "tui_20260828.log" + handle = path.open("w", encoding="utf-8") + self.addCleanup(handle.close) + return path, handle + + class RunStepsInlineTests(unittest.TestCase): def test_runs_steps_in_order_and_returns_zero(self): order = [] @@ -128,10 +148,100 @@ class ProgressParsingTests(_FakeTui, unittest.TestCase): def test_log_tail_is_capped(self): view, _ = self.make_view(steps=[_step("a")]) - for i in range(taskview._LOG_TAIL + 5): + for i in range(taskview._LOG_KEEP + 5): view._ingest_line(f"line {i}") - self.assertEqual(len(view.log_tail), taskview._LOG_TAIL) - self.assertEqual(view.log_tail[-1], f"line {taskview._LOG_TAIL + 4}") + self.assertEqual(len(view.log_tail), taskview._LOG_KEEP) + self.assertEqual(view.log_tail[-1], f"line {taskview._LOG_KEEP + 4}") + + +class ConsoleLogTests(_TempLog, _FakeTui, unittest.TestCase): + """The tui_ day-stream mirror: header, lines, markers, exclusions.""" + + def test_run_header_lines_and_step_markers_are_written(self): + path, handle = self._log_target() + with patch.object(taskview.logging_kit, "day_stream", + return_value=handle): + view, _ = self.make_view(steps=[_step("one")]) + view.handle_event({"kind": "step_start", "index": 0, + "title": "one"}) + view._ingest_line("cloning into 'audio.cpp'...") + view.handle_event({"kind": "step_done", "index": 0, "rc": 0}) + view.handle_event({"kind": "finish", "phase": "done", "rc": 0}) + view._on_stop() + text = path.read_text(encoding="utf-8") + self.assertIn("=== Setup —", text) + self.assertIn("--- one ---", text) + self.assertIn("cloning into 'audio.cpp'...", text) + self.assertIn("[OK] one (exit 0)", text) + self.assertIn("=== done (exit 0) ===", text) + + def test_machine_progress_lines_are_not_logged(self): + path, handle = self._log_target() + with patch.object(taskview.logging_kit, "day_stream", + return_value=handle): + view, _ = self.make_view(steps=[_step("a")]) + view._ingest_line("AUDIOCPP_PROGRESS downloaded=1 total=2") + view._ingest_line("[ 45%] building") + view._on_stop() + text = path.read_text(encoding="utf-8") + self.assertNotIn("AUDIOCPP_PROGRESS", text) + self.assertIn("[ 45%] building", text) + + def test_failed_step_is_marked_in_the_log(self): + path, handle = self._log_target() + with patch.object(taskview.logging_kit, "day_stream", + return_value=handle): + view, _ = self.make_view(steps=[_step("one")]) + view.handle_event({"kind": "step_start", "index": 0, + "title": "one"}) + view.handle_event({"kind": "step_done", "index": 0, "rc": 7}) + view._on_stop() + text = path.read_text(encoding="utf-8") + self.assertIn("[FAIL] one (exit 7)", text) + + def test_cancelled_step_is_marked_in_the_log(self): + path, handle = self._log_target() + with patch.object(taskview.logging_kit, "day_stream", + return_value=handle): + view, _ = self.make_view(steps=[_step("one")]) + view.handle_event({"kind": "step_start", "index": 0, + "title": "one"}) + view.handle_event({"kind": "step_cancelled", "index": 0}) + view.handle_event({"kind": "finish", "phase": "cancelled", + "rc": 1}) + view._on_stop() + text = path.read_text(encoding="utf-8") + self.assertIn("[x] one (cancelled)", text) + self.assertIn("=== cancelled (exit 1) ===", text) + + def test_unwritable_log_disables_the_mirror(self): + with patch.object(taskview.logging_kit, "day_stream", + return_value=None): + view, _ = self.make_view(steps=[_step("a")]) + view._ingest_line("hello") # must not raise + view._on_stop() + self.assertEqual(view.log_tail, ["hello"]) + + def test_lane_output_is_mirrored_with_lane_markers(self): + path, handle = self._log_target() + with patch.object(taskview.logging_kit, "day_stream", + return_value=handle): + screen = FakeScreen(width=80, height=24) + view = taskview.LanesView( + screen, "Setup", + [taskview.TaskLane("Build", [_step("one")])], + clock=lambda: 1000.0) + lane = view._lanes[0] + view._handle_lane_event(lane, {"kind": "step_start", + "index": 0, "title": "one"}) + view._ingest_lane_line(lane, "compiling foo.o") + view._handle_lane_event(lane, {"kind": "lane_finish", "rc": 0}) + view._console_log.close() + text = path.read_text(encoding="utf-8") + self.assertIn("=== Setup —", text) + self.assertIn("--- [Build] one ---", text) + self.assertIn("compiling foo.o", text) + self.assertIn("=== [Build] finished (exit 0) ===", text) class StateTransitionTests(_FakeTui, unittest.TestCase): @@ -193,6 +303,30 @@ class RenderTests(_FakeTui, unittest.TestCase): self.assertIn("two", text) self.assertIn("Esc or q: cancel", text) + def test_log_tail_fills_the_available_height(self): + # A tall terminal shows all 20 lines (the old 10-line cap would + # have left the bottom blank). + view, screen = self.make_view(steps=[_step("one")], height=30) + for i in range(20): + view._ingest_line(f"line {i}") + view.render() + shown = [t for _, _, t, _ in screen.strings if t.startswith("line ")] + self.assertEqual(len(shown), 20) + self.assertEqual(shown[0], "line 0") + self.assertEqual(shown[-1], "line 19") + + def test_log_tail_shows_the_most_recent_lines_that_fit(self): + # The tail is anchored to the latest output: on a small screen the + # earliest lines scroll off while the newest stay visible. + view, screen = self.make_view(steps=[_step("one")], height=24) + for i in range(20): + view._ingest_line(f"line {i}") + view.render() + shown = [t for _, _, t, _ in screen.strings if t.startswith("line ")] + self.assertGreater(len(shown), 10) # more than the old fixed cap + self.assertEqual(shown[-1], "line 19") + self.assertNotIn("line 0", shown) + def test_done_screen_shows_the_completion_footer(self): view, screen = self.make_view(steps=[_step("one")]) view.handle_event({"kind": "step_start", "index": 0, "title": "one"}) -- cgit v1.2.3