aboutsummaryrefslogtreecommitdiff
path: root/app/tests
diff options
context:
space:
mode:
Diffstat (limited to 'app/tests')
-rw-r--r--app/tests/test_hub.py7
-rw-r--r--app/tests/test_logging_kit.py118
-rw-r--r--app/tests/test_runview.py65
-rw-r--r--app/tests/test_taskview.py140
4 files changed, 324 insertions, 6 deletions
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"})
generated by cgit v1.2.3 (git 2.25.1) at 2026-09-08 10:22:51 +0000