aboutsummaryrefslogtreecommitdiff
path: root/app/tests
diff options
context:
space:
mode:
Diffstat (limited to 'app/tests')
-rw-r--r--app/tests/cover_test.pngbin6801 -> 0 bytes
-rw-r--r--app/tests/gen_test_cover.py8
-rw-r--r--app/tests/test_audio.py18
-rw-r--r--app/tests/test_audiobook_cli.py5
-rw-r--r--app/tests/test_backends_audiocpp.py74
-rw-r--r--app/tests/test_backends_common.py28
-rw-r--r--app/tests/test_backends_servers.py41
-rw-r--r--app/tests/test_converter.py6
-rw-r--r--app/tests/test_converter_progress.py18
-rw-r--r--app/tests/test_cover.py30
-rw-r--r--app/tests/test_extractors.py80
-rw-r--r--app/tests/test_runview.py11
-rw-r--r--app/tests/test_taskview.py6
-rw-r--r--app/tests/test_tts.py279
-rw-r--r--app/tests/test_viewkit.py9
15 files changed, 365 insertions, 248 deletions
diff --git a/app/tests/cover_test.png b/app/tests/cover_test.png
deleted file mode 100644
index 0c252db..0000000
--- a/app/tests/cover_test.png
+++ /dev/null
Binary files differ
diff --git a/app/tests/gen_test_cover.py b/app/tests/gen_test_cover.py
deleted file mode 100644
index 292469a..0000000
--- a/app/tests/gen_test_cover.py
+++ /dev/null
@@ -1,8 +0,0 @@
-import sys
-from pathlib import Path
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-from converter.cover import generate_cover
-
-p = generate_cover('The Count of Monte Cristo',
- Path(__file__).resolve().parent / 'cover_test.png')
-print('written:', p)
diff --git a/app/tests/test_audio.py b/app/tests/test_audio.py
index 6c67294..75de97c 100644
--- a/app/tests/test_audio.py
+++ b/app/tests/test_audio.py
@@ -22,34 +22,34 @@ from converter.audio import (
build_m4b_chapters_command,
cleanup_chunks,
concat_audio_files,
- speed_export_params,
+ atempo_filters,
verify_output_duration,
)
-class SpeedExportParamsTests(unittest.TestCase):
+class AtempoFiltersTests(unittest.TestCase):
def test_normal_speed_no_filter(self):
- self.assertEqual(speed_export_params(1.0), [])
+ self.assertEqual(atempo_filters(1.0), "")
def test_simple_speedup(self):
- self.assertEqual(speed_export_params(1.5), ["-filter:a", "atempo=1.5"])
+ self.assertEqual(atempo_filters(1.5), "atempo=1.5")
def test_simple_slowdown(self):
- self.assertEqual(speed_export_params(0.75), ["-filter:a", "atempo=0.75"])
+ self.assertEqual(atempo_filters(0.75), "atempo=0.75")
def test_chained_speedup_beyond_2x(self):
- self.assertEqual(speed_export_params(3.0), ["-filter:a", "atempo=2.0,atempo=1.5"])
+ self.assertEqual(atempo_filters(3.0), "atempo=2.0,atempo=1.5")
def test_chained_slowdown_below_half(self):
- self.assertEqual(speed_export_params(0.25), ["-filter:a", "atempo=0.5,atempo=0.5"])
+ self.assertEqual(atempo_filters(0.25), "atempo=0.5,atempo=0.5")
def test_zero_speed_rejected(self):
with self.assertRaises(ValueError):
- speed_export_params(0)
+ atempo_filters(0)
def test_negative_speed_rejected(self):
with self.assertRaises(ValueError):
- speed_export_params(-1.5)
+ atempo_filters(-1.5)
class CleanupChunksTests(unittest.TestCase):
diff --git a/app/tests/test_audiobook_cli.py b/app/tests/test_audiobook_cli.py
index f1eb8e8..8076c8e 100644
--- a/app/tests/test_audiobook_cli.py
+++ b/app/tests/test_audiobook_cli.py
@@ -390,12 +390,16 @@ class AllModelsConvertTests(unittest.TestCase):
ctor_kwargs.append(ckwargs)
inst = MagicMock()
made.append(inst)
+ result = True
if make_run is not None:
inst.run.side_effect = make_run(ckwargs)
else:
result = states[len(made) - 1] \
if len(made) <= len(states) else True
inst.run.return_value = result
+ # Per-book outcomes the All-run loop counts on the CLI path
+ # (see audiobook._convert_each_model).
+ inst.results = {"book.txt": bool(result)}
return inst
fake_class = MagicMock(side_effect=make_instance)
@@ -550,6 +554,7 @@ class AllModelsConvertTests(unittest.TestCase):
def make_instance(*args, **ckwargs):
inst = MagicMock()
inst.run.return_value = True
+ inst.results = {"book.txt": True}
return inst
fake_class.side_effect = make_instance
with patch.object(audiobook, "setup_logging"), \
diff --git a/app/tests/test_backends_audiocpp.py b/app/tests/test_backends_audiocpp.py
index 18c38c4..9392ba1 100644
--- a/app/tests/test_backends_audiocpp.py
+++ b/app/tests/test_backends_audiocpp.py
@@ -410,6 +410,56 @@ class LoadModelCatalogTests(unittest.TestCase):
self.assertIn("mystery_tts",
[entry["family"] for entry in catalog])
+ def test_non_object_spec_json_is_skipped_not_fatal(self):
+ # Valid JSON that is not an object (the "crash on bad model_specs"
+ # fix class): skipped like an unparsable spec, never an
+ # AttributeError out of the wizard.
+ for payload in ('["a list"]', '"a string"', "42", "null"):
+ (self.checkout / "model_specs" / "broken.json") \
+ .write_text(payload, encoding="utf-8")
+ catalog = make_server.catalog.load_model_catalog(self.checkout)
+ self.assertNotIn("broken",
+ [entry["family"] for entry in catalog])
+
+ def test_non_dict_package_entries_are_skipped(self):
+ (self.checkout / "model_specs" / "weird_pkg.json").write_text(
+ json.dumps({"family": "weird_pkg", "tasks": ["tts"],
+ "packages": ["not-a-dict",
+ {"id": "wp", "format": "gguf",
+ "files": ["m.gguf"],
+ "target_directory": "weird_pkg"}]}),
+ encoding="utf-8")
+ catalog = make_server.catalog.load_model_catalog(self.checkout)
+ entry = next(e for e in catalog if e["family"] == "weird_pkg")
+ self.assertEqual(entry["install_id"], "wp")
+
+ def test_download_path_materializes_the_catalog_repair(self):
+ # Regression for the shadowed-sanitizer bug: the download path's
+ # staged specs copy must apply the catalog sanitizer's SECOND bug
+ # class (missing strip_prefix on $gguf-rooted single-GGUF packages
+ # nested under a repo directory — glm_tts/outetts), which the old
+ # dot-only models.py repair did not.
+ (self.checkout / "model_specs" / "glm_like.json").write_text(
+ json.dumps({"family": "glm_like", "tasks": ["tts"],
+ "sources": [{"format": "gguf",
+ "roots": {"tokenizer": "$gguf"}}],
+ "packages": [{"id": "glm_q8", "format": "gguf",
+ "files": ["Text to audio (TTS)/"
+ "GLM-TTS_Q8.gguf"]},
+ {"id": "glm_other", "format": "safetensors",
+ "files": ["tokenizer_merges"],
+ "default": True}]}),
+ encoding="utf-8")
+ staging = make_server.models._prepare_specs_dir(self.checkout)
+ self.assertIsNotNone(staging, "nested-GGUF repair was not staged")
+ try:
+ repaired = json.loads(
+ (staging / "glm_like.json").read_text(encoding="utf-8"))
+ finally:
+ shutil.rmtree(staging, ignore_errors=True)
+ self.assertEqual(repaired["packages"][0]["strip_prefix"],
+ "Text to audio (TTS)")
+
def test_families_sorted_alphabetically_by_display_name(self):
catalog = make_server.catalog.load_model_catalog(self.checkout)
names = [entry["display_name"].lower() for entry in catalog]
@@ -1095,48 +1145,53 @@ class InstallModelsTests(unittest.TestCase):
class SanitizeModelSpecTests(unittest.TestCase):
- """The dot strip_prefix repair and the --specs-dir staging copy."""
+ """The strip_prefix repairs (both upstream bug classes) and staging."""
def test_dot_prefix_dropped_when_file_is_bare(self):
spec = {"packages": [{"files": ["model.gguf"], "strip_prefix": "."}]}
- self.assertTrue(make_server.models._sanitize_model_spec(spec))
+ self.assertTrue(make_server.catalog.sanitize_model_spec(spec))
self.assertEqual(spec["packages"][0]["strip_prefix"], "")
def test_slash_dot_prefix_normalized_like_dot(self):
spec = {"packages": [{"files": ["model.gguf"], "strip_prefix": "./"}]}
- self.assertTrue(make_server.models._sanitize_model_spec(spec))
+ self.assertTrue(make_server.catalog.sanitize_model_spec(spec))
self.assertEqual(spec["packages"][0]["strip_prefix"], "")
def test_dot_prefix_kept_when_files_carry_it(self):
spec = {"packages": [{"files": ["./model.gguf"],
"strip_prefix": "."}]}
- self.assertFalse(make_server.models._sanitize_model_spec(spec))
+ self.assertFalse(make_server.catalog.sanitize_model_spec(spec))
self.assertEqual(spec["packages"][0]["strip_prefix"], ".")
def test_real_directory_prefix_untouched(self):
spec = {"packages": [{"files": ["model.gguf"],
"strip_prefix": "Kroko-ASR-GGUF"}]}
- self.assertFalse(make_server.models._sanitize_model_spec(spec))
+ self.assertFalse(make_server.catalog.sanitize_model_spec(spec))
self.assertEqual(spec["packages"][0]["strip_prefix"],
"Kroko-ASR-GGUF")
def test_valid_prefix_untouched(self):
spec = {"packages": [{"files": ["Kroko-ASR-GGUF/model.gguf"],
"strip_prefix": "Kroko-ASR-GGUF"}]}
- self.assertFalse(make_server.models._sanitize_model_spec(spec))
+ self.assertFalse(make_server.catalog.sanitize_model_spec(spec))
- def test_missing_or_empty_files_untouched(self):
+ def test_malformed_files_package_repaired_too(self):
+ # The catalog sanitizer also drops a dot prefix when the package's
+ # files list is missing, empty, or malformed — nothing can match a
+ # dot prefix, so the repair is safe there as well (a dot prefix
+ # with files that all carry it is kept, see above).
spec = {"packages": [{"strip_prefix": "."},
{"files": [], "strip_prefix": "."},
{"files": "model.gguf", "strip_prefix": "."}]}
- self.assertFalse(make_server.models._sanitize_model_spec(spec))
+ self.assertTrue(make_server.catalog.sanitize_model_spec(spec))
+ self.assertEqual({p["strip_prefix"] for p in spec["packages"]}, {""})
def test_only_broken_packages_repaired(self):
spec = {"packages": [
{"files": ["model.gguf"], "strip_prefix": "."},
{"files": ["./model.gguf"], "strip_prefix": "."},
]}
- self.assertTrue(make_server.models._sanitize_model_spec(spec))
+ self.assertTrue(make_server.catalog.sanitize_model_spec(spec))
self.assertEqual([p["strip_prefix"] for p in spec["packages"]],
["", "."])
@@ -3368,7 +3423,6 @@ class ExecuteLanesTests(unittest.TestCase):
"include_clone": False,
"wav_dir": None,
"plan": None,
- "sync_port": None,
"delete_unused": False,
"unused_entries": [],
"model_entries": [],
diff --git a/app/tests/test_backends_common.py b/app/tests/test_backends_common.py
index 08ffc2c..68a5543 100644
--- a/app/tests/test_backends_common.py
+++ b/app/tests/test_backends_common.py
@@ -51,6 +51,34 @@ class RunConsoleSubprocessStreamingTests(unittest.TestCase):
on_cancel=lambda: touched.append(True))
self.assertEqual(touched, [True])
+ def test_carriage_return_progress_streams_incrementally(self):
+ # tqdm/HuggingFace-style \r-only progress: a readline-based reader
+ # blocked until the next \n, so the updates arrived in one burst
+ # (or the stall watchdog fired first). Each \r segment must be
+ # emitted as its own line.
+ lines = []
+ rc = common.run_console_subprocess(
+ [sys.executable, "-c",
+ "import sys, time\n"
+ "for i in range(4):\n"
+ " sys.stdout.write(f'pct {i}\\r'); sys.stdout.flush()\n"
+ " time.sleep(0.2)\n"
+ "sys.stdout.write('done\\n'); sys.stdout.flush()\n"],
+ emit=lines.append, stall_timeout=1.0)
+ self.assertEqual(rc, 0)
+ self.assertEqual(lines, [f"pct {i}" for i in range(4)] + ["done"])
+
+ def test_url_with_port_preserves_userinfo_and_ipv6(self):
+ self.assertEqual(
+ common.url_with_port("http://user:pass@host:8000", 8080),
+ "http://user:pass@host:8080")
+ self.assertEqual(
+ common.url_with_port("http://[::1]:8000", 8080),
+ "http://[::1]:8080")
+ self.assertEqual(
+ common.url_with_port("http://host:8000/path", 8080),
+ "http://host:8080/path")
+
class RunConsoleSubprocessStallTests(unittest.TestCase):
"""The no-output watchdog: a silent child is killed and reported 124."""
diff --git a/app/tests/test_backends_servers.py b/app/tests/test_backends_servers.py
index ab05eed..61897d4 100644
--- a/app/tests/test_backends_servers.py
+++ b/app/tests/test_backends_servers.py
@@ -65,10 +65,45 @@ class StartTests(unittest.TestCase):
ok = servers.start(self.spec)
self.assertTrue(ok)
mk.assert_called_once()
- # Pid file written.
+ # Pid file written (first field is the pid; the optional second
+ # field is the start-time ownership token, absent on this platform).
self.assertEqual(
- (self.dir / "test-server.pid").read_text(encoding="utf-8"),
- "4242")
+ (self.dir / "test-server.pid").read_text(encoding="utf-8")
+ .split()[0], "4242")
+
+ def test_pid_file_records_a_start_time_token_where_available(self):
+ proc = MagicMock()
+ proc.pid = 5150
+ proc.poll.return_value = None
+ with patch.object(servers, "LOG_DIR", self.dir), \
+ patch("subprocess.Popen", return_value=proc), \
+ patch.object(servers, "_process_start_token",
+ return_value="12345"), \
+ patch("backends.common.server_running",
+ side_effect=[False, True]), \
+ patch("time.sleep"):
+ self.assertTrue(servers.start(self.spec))
+ fields = (self.dir / "test-server.pid") \
+ .read_text(encoding="utf-8").split()
+ self.assertEqual(fields, ["5150", "12345"])
+
+ def test_recycled_pid_with_mismatched_token_is_not_ours(self):
+ # The pid is alive but its start time differs from the recorded
+ # token: an unrelated process now owns this pid, so manages/alive
+ # must report not-ours (and never kill it).
+ pid_file = self.dir / "test-server.pid"
+ pid_file.write_text("4242 111\n", encoding="utf-8")
+ with patch.object(servers, "LOG_DIR", self.dir), \
+ patch.object(servers, "_pid_alive", return_value=True), \
+ patch.object(servers, "_process_start_token",
+ return_value="999"):
+ self.assertFalse(servers.alive("test"))
+ self.assertFalse(servers.manages([self.spec]))
+ with patch.object(servers, "LOG_DIR", self.dir), \
+ patch.object(servers, "_pid_alive", return_value=True), \
+ patch.object(servers, "_process_start_token",
+ return_value="111"):
+ self.assertTrue(servers.alive("test"))
def test_returns_false_when_process_exits_early(self):
proc = MagicMock()
diff --git a/app/tests/test_converter.py b/app/tests/test_converter.py
index 41b9cad..0b0eb0d 100644
--- a/app/tests/test_converter.py
+++ b/app/tests/test_converter.py
@@ -38,6 +38,12 @@ class SanitizeFilenameTests(unittest.TestCase):
def test_empty_falls_back(self):
self.assertEqual(AudiobookConverter._sanitize_filename("///"), "chapter")
+ def test_windows_reserved_device_names_are_suffixed(self):
+ for name in ("CON", "nul", "COM1", "lpt2"):
+ result = AudiobookConverter._sanitize_filename(name)
+ self.assertNotEqual(result.upper(), name.upper())
+ self.assertTrue(result.startswith("chapter_"), result)
+
class ConfigurationValidationTests(unittest.TestCase):
def test_invalid_voice_mode_rejected(self):
diff --git a/app/tests/test_converter_progress.py b/app/tests/test_converter_progress.py
index d77e9e0..0a44725 100644
--- a/app/tests/test_converter_progress.py
+++ b/app/tests/test_converter_progress.py
@@ -225,6 +225,24 @@ class CancelTests(unittest.TestCase):
converter = self.fixture.build(cancel=threading.Event())
converter._check_cancelled() # no raise
+ def test_cancel_event_reaches_the_client_before_connect(self):
+ # The cancel event must be wired into the TTS client at
+ # construction time — connect-time work (health check, model
+ # listing, voice validation) is otherwise un-cancellable.
+ cancel = threading.Event()
+ cancel.set()
+ client = MagicMock()
+ client.cancel = None
+ with patch.object(converter_mod, "QwenTTSClient",
+ return_value=client) as mock_qwen:
+ AudiobookConverter(voice_mode=VOICE_MODE_CUSTOM,
+ backend=BACKEND_QWEN, voice="Vivian",
+ cancel=cancel)
+ self.assertIs(mock_qwen.call_args.kwargs["cancel"], cancel)
+ with self.assertRaises(ConversionCancelled):
+ converter_mod.AudiobookConverter._check_cancelled(
+ MagicMock(tts=None, _cancel=cancel))
+
if __name__ == "__main__":
unittest.main()
diff --git a/app/tests/test_cover.py b/app/tests/test_cover.py
index f19db5a..ca8b9a2 100644
--- a/app/tests/test_cover.py
+++ b/app/tests/test_cover.py
@@ -9,6 +9,7 @@ from pathlib import Path
from converter.cover import (
_random_light_color,
+ _render_line,
_text_width,
_wrap_title,
generate_cover,
@@ -117,11 +118,13 @@ class GenerateCoverTests(unittest.TestCase):
_, _, rows = _decode_png(data)
self.assertEqual(_black_pixels(rows), 0)
- def test_unrenderable_title_degrades_to_gradient(self):
- # CJK glyphs are not in the bitmap font; no crash, no text pixels
+ def test_unrenderable_title_renders_placeholder_boxes(self):
+ # CJK glyphs are not in the bitmap font: they used to vanish
+ # (blank cover); now each renders as a black-outlined placeholder
+ # box so the title is visibly present.
_, data = self._write("书名")
_, _, rows = _decode_png(data)
- self.assertEqual(_black_pixels(rows), 0)
+ self.assertGreater(_black_pixels(rows), 0)
def test_write_failure_returns_none(self):
result = generate_cover("Hello", Path("/nonexistent_dir/cover.png"))
@@ -187,3 +190,24 @@ class DropShadowTests(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
+
+
+class NonAsciiTitleTests(unittest.TestCase):
+ """Characters outside the bitmap font render as boxes, not blanks."""
+
+ def test_unknown_characters_are_rendered_as_placeholder_boxes(self):
+ # Regression: unknown chars used to be skipped in rendering while
+ # still counted for width, so e.g. Japanese titles produced a
+ # blank cover with no warning.
+ with tempfile.TemporaryDirectory() as tmp:
+ path = generate_cover("日本語", Path(tmp) / "c.png",
+ seed=1)
+ self.assertIsNotNone(path)
+ raw = path.read_bytes()
+ self.assertTrue(raw.startswith(b"\x89PNG"))
+
+ def test_placeholder_draws_pixels_for_every_unknown_char(self):
+ pixels = [[(255, 255, 255)] * 60 for _ in range(60)]
+ _render_line(pixels, "éé", 2, 2, color=(0, 0, 0))
+ dark = sum(1 for row in pixels for pixel in row if pixel == (0, 0, 0))
+ self.assertGreater(dark, 2 * 5 * 7 - 6) # both boxes' outlines drawn
diff --git a/app/tests/test_extractors.py b/app/tests/test_extractors.py
index 64666ba..619fe5a 100644
--- a/app/tests/test_extractors.py
+++ b/app/tests/test_extractors.py
@@ -53,6 +53,86 @@ class TxtExtractionTests(unittest.TestCase):
with self.assertRaises(ValueError):
extract_text(path)
+ def test_utf16_without_bom_detected(self):
+ self.assertEqual(self._extract("chapter one".encode("utf-16-le")),
+ "chapter one")
+
+ def test_lone_nul_does_not_flip_to_utf16(self):
+ # A single stray NUL byte in an otherwise-ASCII UTF-8 file must not
+ # switch the whole book to a UTF-16 decode (mojibake): the text
+ # comes back readable instead.
+ self.assertEqual(self._extract(b"hello world\x00rest"),
+ "hello world\x00rest")
+
+ def test_standalone_page_numbers_removed_but_years_kept(self):
+ from converter.extractors import clean_text
+
+ cleaned = clean_text("Chapter 1\n\n42\n\nIt was 1984.")
+ self.assertNotIn("42", cleaned)
+ self.assertIn("1984", cleaned)
+ kept = clean_text("It was the year\n\n1984\n\nwhen it began.")
+ self.assertIn("1984", kept)
+
+
+class EpubZipfileFallbackTests(unittest.TestCase):
+ """The no-ebooklib EPUB fallback: spine order, no TOC narration."""
+
+ @staticmethod
+ def _write_epub(path: Path):
+ import zipfile
+
+ container = ("<?xml version=\"1.0\"?>"
+ "<container><rootfiles>"
+ "<rootfile full-path=\"OEBPS/content.opf\"/>"
+ "</rootfiles></container>")
+ opf = ("<?xml version=\"1.0\"?>"
+ "<package xmlns=\"http://www.idpf.org/2007/opf\">"
+ "<manifest>"
+ "<item id=\"nav\" href=\"nav.xhtml\" properties=\"nav\"/>"
+ "<item id=\"c2\" href=\"text/chapterB.xhtml\"/>"
+ "<item id=\"c1\" href=\"text/chapterA.xhtml\"/>"
+ "</manifest>"
+ "<spine><itemref idref=\"nav\"/>"
+ "<itemref idref=\"c2\"/><itemref idref=\"c1\"/></spine>"
+ "</package>")
+ with zipfile.ZipFile(path, "w") as zf:
+ zf.writestr("mimetype", "application/epub+zip")
+ zf.writestr("META-INF/container.xml", container)
+ zf.writestr("OEBPS/content.opf", opf)
+ zf.writestr("OEBPS/nav.xhtml",
+ "<html><body><p>Contents</p></body></html>")
+ zf.writestr("OEBPS/text/chapterA.xhtml",
+ "<html><body><p>Alpha text.</p></body></html>")
+ zf.writestr("OEBPS/text/chapterB.xhtml",
+ "<html><body><p>Beta text.</p></body></html>")
+
+ def test_spine_order_and_no_nav(self):
+ from converter.extractors import _read_epub_zipfile
+
+ with tempfile.TemporaryDirectory() as tmp:
+ path = Path(tmp) / "book.epub"
+ self._write_epub(path)
+ items = _read_epub_zipfile(path)
+
+ titles = [title for title, _ in items]
+ self.assertNotIn("nav", titles)
+ # Spine order (B before A) beats filename sort (A before B).
+ self.assertEqual(titles, ["chapterB", "chapterA"])
+
+ def test_unparsable_opf_falls_back_to_filename_order(self):
+ import zipfile
+
+ from converter.extractors import _read_epub_zipfile
+
+ with tempfile.TemporaryDirectory() as tmp:
+ path = Path(tmp) / "book.epub"
+ with zipfile.ZipFile(path, "w") as zf:
+ zf.writestr("a.xhtml", "<html><body><p>A</p></body></html>")
+ zf.writestr("b.xhtml", "<html><body><p>B</p></body></html>")
+ items = _read_epub_zipfile(path)
+
+ self.assertEqual([title for title, _ in items], ["a", "b"])
+
def _build_test_epub(path: Path, chapters=(("One", "First chapter text."),
("Two", "Second chapter text."))) -> None:
diff --git a/app/tests/test_runview.py b/app/tests/test_runview.py
index 4e3f9fe..6add83f 100644
--- a/app/tests/test_runview.py
+++ b/app/tests/test_runview.py
@@ -277,6 +277,17 @@ class LogAppenderTests(_FakeTui, unittest.TestCase):
self.assertTrue(lines[0].endswith(" - one"))
self.assertTrue(lines[2].endswith(" - three"))
+ def test_carriage_return_progress_splits_into_lines(self):
+ # tqdm/git-style \r-only progress: each segment becomes its own
+ # log line instead of one ever-growing buffered line.
+ path, appender = self._appender()
+ appender.write("pct 0\rpct 1\rpct 2\r")
+ appender.flush()
+ with open(path, encoding="utf-8") as handle:
+ lines = handle.read().splitlines()
+ self.assertEqual(len(lines), 3)
+ self.assertTrue(lines[2].endswith(" - pct 2"))
+
def test_blank_lines_and_empty_path_are_skipped(self):
path, appender = self._appender()
appender.write("\n\n")
diff --git a/app/tests/test_taskview.py b/app/tests/test_taskview.py
index 78cc2f1..2b3e542 100644
--- a/app/tests/test_taskview.py
+++ b/app/tests/test_taskview.py
@@ -277,7 +277,9 @@ class StateTransitionTests(_FakeTui, unittest.TestCase):
view.handle_event({"kind": "step_cancelled", "index": 0})
view.handle_event({"kind": "finish", "phase": "cancelled", "rc": 1})
self.assertEqual(view.phase, "cancelled")
- self.assertEqual(view._result_rc(), 1)
+ # 130: the CLI's user-cancel code, so callers can flash
+ # "cancelled" instead of "failed" (see hub._download_models_action).
+ self.assertEqual(view._result_rc(), 130)
# The step interrupted by cancel is marked cancelled, not failed.
self.assertEqual(view._step_mark(0), ("[x]", "warn"))
self.assertEqual(view._step_mark(1), ("[ ]", "dim"))
@@ -677,7 +679,7 @@ class LanesViewTests(_FakeTui, unittest.TestCase):
view._drain()
self.assertEqual(view.phase, "cancelled")
self.assertTrue(view.cancelled)
- self.assertEqual(view._result_rc(), 1)
+ self.assertEqual(view._result_rc(), 130)
def test_split_render_draws_both_lane_titles(self):
view, screen = self.make_view(self._two_lanes())
diff --git a/app/tests/test_tts.py b/app/tests/test_tts.py
index 45e3812..39b407d 100644
--- a/app/tests/test_tts.py
+++ b/app/tests/test_tts.py
@@ -41,14 +41,15 @@ from converter.clients import (
VOICE_MODES,
AudioCppTTSClient,
FasterTTSClient,
+ QWEN3_TTS_SPEAKERS,
QwenTTSClient,
audiocpp_entry_voice_capability,
audiocpp_family_narrates,
audiocpp_family_voice_policy,
audiocpp_request_error,
audiocpp_script_input,
+ audiocpp_voice_for_run,
allocation_log_note,
- build_trimmed_voice_reference,
nvidia_device_memory_report,
normalize_language,
transcribe_reference_audio_detailed,
@@ -2310,7 +2311,7 @@ class BackendWiringTests(unittest.TestCase):
backend=BACKEND_FASTER, voice="narrator")
mock_faster.assert_called_once_with(chunks_dir=converter_mod.CHUNKS_FOLDER,
voice="narrator", api_url=None,
- quiet=False)
+ quiet=False, cancel=None)
mock_qwen.assert_not_called()
mock_audiocpp.assert_not_called()
@@ -2327,7 +2328,7 @@ class BackendWiringTests(unittest.TestCase):
instructions=None,
request_options={},
api_url=None, quiet=False,
- unload_models=None)
+ unload_models=None, cancel=None)
mock_faster.assert_not_called()
mock_qwen.assert_not_called()
@@ -2341,7 +2342,7 @@ class BackendWiringTests(unittest.TestCase):
instructions=None,
request_options={},
api_url=None, quiet=False,
- unload_models=None)
+ unload_models=None, cancel=None)
def test_audiocpp_backend_model_id_is_wired_through(self):
with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp:
@@ -2353,7 +2354,7 @@ class BackendWiringTests(unittest.TestCase):
voice="narrator", language=config.LANGUAGE,
model_id="higgs", instructions=None,
request_options={}, api_url=None, quiet=False,
- unload_models=None)
+ unload_models=None, cancel=None)
def test_audiocpp_backend_instructions_and_options_are_wired_through(self):
with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp:
@@ -2368,7 +2369,8 @@ class BackendWiringTests(unittest.TestCase):
model_id=None,
instructions="A warm adult narrator",
request_options={"emotion": "neutral", "speed": "1.1"},
- api_url=None, quiet=False, unload_models=None)
+ api_url=None, quiet=False, unload_models=None,
+ cancel=None)
def test_qwen_backend_uses_qwen_client(self):
with patch("converter.converter.FasterTTSClient") as mock_faster, \
@@ -2420,7 +2422,7 @@ class BackendWiringTests(unittest.TestCase):
voice="narrator", language=config.LANGUAGE, model_id=None,
instructions=None, request_options={},
api_url="http://10.0.0.5:8080", quiet=False,
- unload_models=None)
+ unload_models=None, cancel=None)
with patch("converter.converter.FasterTTSClient") as mock_faster:
AudiobookConverter(voice_mode=VOICE_MODE_CLONE,
backend=BACKEND_FASTER, voice="narrator",
@@ -2428,7 +2430,7 @@ class BackendWiringTests(unittest.TestCase):
mock_faster.assert_called_once_with(chunks_dir=converter_mod.CHUNKS_FOLDER,
voice="narrator",
api_url="http://10.0.0.5:8000",
- quiet=False)
+ quiet=False, cancel=None)
with patch("converter.converter.QwenTTSClient") as mock_qwen:
AudiobookConverter(voice_mode=VOICE_MODE_CUSTOM,
backend=BACKEND_QWEN, voice="Vivian",
@@ -2438,7 +2440,8 @@ class BackendWiringTests(unittest.TestCase):
voice_mode=VOICE_MODE_CUSTOM, voice_clone_ref_audio=None,
voice_clone_ref_text=None, skip_transcription=False,
language=config.LANGUAGE, instructions=None,
- api_url="http://10.0.0.5:7860", quiet=False, voice="Vivian")
+ api_url="http://10.0.0.5:7860", quiet=False, voice="Vivian",
+ cancel=None)
def test_audiocpp_clone_mode_does_not_require_reference(self):
# Cloning is server-side for the audiocpp backend, so the
@@ -2548,77 +2551,6 @@ if __name__ == "__main__":
unittest.main()
-class TrimmedVoiceReferenceTests(unittest.TestCase):
- """build_trimmed_voice_reference: a bounded inline cloning reference."""
-
- def setUp(self):
- self._tmp = tempfile.TemporaryDirectory()
- self.dir = Path(self._tmp.name)
-
- def tearDown(self):
- self._tmp.cleanup()
-
- @staticmethod
- def _wav_bytes(rate, channels, sampwidth, frames, fill):
- if sampwidth == 1:
- payload = bytes(fill & 0xFF for _ in range(frames * channels))
- else:
- payload = fill.to_bytes(sampwidth, "little", signed=True) \
- * frames * channels
- return (b"RIFF" + struct.pack("<I", 36 + len(payload)) + b"WAVEfmt "
- + struct.pack("<IHHIIHH", 16, 1, channels, rate,
- rate * channels * sampwidth,
- channels * sampwidth, sampwidth * 8)
- + b"data" + struct.pack("<I", len(payload)) + payload)
-
- def _write(self, name, rate, channels, sampwidth, seconds, fill=256):
- path = self.dir / name
- path.write_bytes(self._wav_bytes(rate, channels, sampwidth,
- int(rate * seconds), fill))
- return path
-
- def test_sixty_second_stereo_reference_is_cut_to_thirty_seconds(self):
- path = self._write("ref.wav", 44100, 2, 2, 60)
- b64, seconds, name = build_trimmed_voice_reference(path)
- self.assertEqual(name, "ref.wav")
- self.assertLessEqual(seconds, 30.0 + 1e-6)
- decoded = base64.b64decode(b64)
- with wave.open(io.BytesIO(decoded)) as handle:
- self.assertEqual(handle.getframerate(), 44100)
- self.assertEqual(handle.getnchannels(), 1)
- self.assertEqual(handle.getsampwidth(), 2)
- self.assertLessEqual(handle.getnframes() / handle.getframerate(),
- 30.0 + 1e-6)
- self.assertLessEqual(len(b64), (5 * 1024 * 1024 + 2) // 3 * 4)
-
- def test_short_reference_is_sent_whole(self):
- path = self._write("short.wav", 16000, 1, 2, 8)
- _b64, seconds, _name = build_trimmed_voice_reference(path)
- self.assertAlmostEqual(seconds, 8.0, places=2)
-
- def test_float_and_garbage_files_yield_none(self):
- self.assertIsNone(build_trimmed_voice_reference(None))
- garbage = self.dir / "garbage.wav"
- garbage.write_bytes(b"ID3 not a wav")
- self.assertIsNone(build_trimmed_voice_reference(garbage))
- missing = self.dir / "missing.wav"
- self.assertIsNone(build_trimmed_voice_reference(missing))
-
- def test_sample_widths_beyond_s16_are_downconverted(self):
- for name, width, seconds in (("s24.wav", 3, 40), ("s32.wav", 4, 40),
- ("u8.wav", 1, 12)):
- path = self._write(name, 44100, 1, width, seconds)
- result = build_trimmed_voice_reference(path)
- self.assertIsNotNone(result, name)
- _b64, used, _name = result
- self.assertLessEqual(used, min(seconds, 30.0) + 1e-6)
-
- def test_base64_payload_stays_under_the_server_limit(self):
- path = self._write("long.wav", 48000, 2, 2, 120)
- b64, _seconds, _name = build_trimmed_voice_reference(path)
- self.assertLessEqual(len(base64.b64decode(b64)), 5 * 1024 * 1024)
-
-
class AllocationLogNoteTests(unittest.TestCase):
"""allocation_log_note: the server log's exact allocation numbers."""
@@ -2695,136 +2627,57 @@ class DeviceMemoryWarningTests(unittest.TestCase):
url="http://10.0.0.5:8080"), "")
-class TrimmedReferenceRetryTests(unittest.TestCase):
- """One trimmed-reference retry after an allocation-failure 500."""
-
- def setUp(self):
- self._tmp = tempfile.TemporaryDirectory()
- self.dir = Path(self._tmp.name)
- frames = 44100 * 60
- payload = (3000).to_bytes(2, "little", signed=True) * frames * 2
- wav = (b"RIFF" + struct.pack("<I", 36 + len(payload)) + b"WAVEfmt "
- + struct.pack("<IHHIIHH", 16, 1, 2, 44100, 44100 * 2, 2, 16)
- + b"data" + struct.pack("<I", len(payload)) + payload)
- (self.dir / "obama.wav").write_bytes(wav)
-
- def tearDown(self):
- self._tmp.cleanup()
-
- _WAV = b"RIFF\x04\x00\x00\x00WAVE"
-
- def _client(self, responses, captured, voice="obama",
- family="moss_tts_local"):
- client = AudioCppTTSClient.__new__(AudioCppTTSClient)
- client.chunks_dir = self.dir
- client.api_url = "http://127.0.0.1:8080"
- client.model_id = "MOSS-TTS-Local-v1.5-GGUF"
- client.preset_mode = True
- client.voice = voice
- client.language = "Auto"
- client._seed = 42
- client.family = family
- client.task = "tts"
- client.profile = audiocpp_client.AUDIOCPP_DEFAULT_FAMILY_PROFILE
- client.instructions = ""
- client.request_options = {}
- client.design_mode = False
- client.instruction_voice = False
- client.plain_mode = False
- client._reference_trim_attempted = False
- client._voice_ref_b64 = None
- client._voice_ref_reference_text = None
- client._voice_wav_path = lambda: self.dir / "obama.wav"
-
- calls = {"n": 0}
-
- def urlopen(request, **_kwargs):
- index = calls["n"]
- calls["n"] += 1
- if captured is not None:
- captured.append(json.loads(request.data.decode("utf-8")))
- outcome = responses[index] if index < len(responses) \
- else responses[-1]
- if isinstance(outcome, urllib.error.HTTPError):
- raise outcome
- response = MagicMock()
- response.__enter__.return_value = response
- response.read.return_value = outcome
- return response
-
- patcher = patch("converter.clients.faster.urllib.request.urlopen",
- side_effect=urlopen)
- patcher.start()
- self.addCleanup(patcher.stop)
- return client
-
- @staticmethod
- def _alloc_http_error():
- return urllib.error.HTTPError(
- "http://127.0.0.1:8080", 500, "Internal Server Error", {},
- io.BytesIO(json.dumps({"error": {"message":
- "failed to allocate MOSS codec encoder forward graph"}}
- ).encode("utf-8")))
-
- def test_allocation_failure_retries_once_with_trimmed_reference(self):
- captured = []
- client = self._client([self._alloc_http_error(), self._WAV, self._WAV],
- captured)
- client._request_wav("Hello.")
- client._request_wav("More text.")
- self.assertEqual(len(captured), 3)
- self.assertIn("voice", captured[0])
- self.assertNotIn("voice_ref", captured[0])
- self.assertIn("voice_ref", captured[1])
- self.assertEqual(captured[1]["voice_ref"]["type"], "base64")
- self.assertLessEqual(
- len(base64.b64decode(captured[1]["voice_ref"]["data"])),
- 5 * 1024 * 1024)
- self.assertNotIn("voice", captured[1])
- # The trimmed reference sticks for the rest of the run.
- self.assertEqual(captured[2]["voice_ref"]["type"], "base64")
- self.assertTrue(client._reference_trim_attempted)
-
- def test_no_local_wav_falls_through_to_the_error(self):
- client = self._client([self._alloc_http_error()], None)
- client._voice_wav_path = lambda: None
- with self.assertRaises(NonRetryableTTSError):
- client._request_wav("Hello.")
- self.assertTrue(client._reference_trim_attempted)
-
- def test_unrelated_errors_are_not_retried(self):
- captured = []
- boring = urllib.error.HTTPError(
- "http://127.0.0.1:8080", 500, "Internal Server Error", {},
- io.BytesIO(json.dumps({"error": {"message": "model busy"}}
- ).encode("utf-8")))
- client = self._client([boring], captured)
- with self.assertRaises(RuntimeError):
- client._request_wav("Hello.")
- self.assertEqual(len(captured), 1)
- self.assertIn("voice", captured[0])
- self.assertFalse(client._reference_trim_attempted)
-
- def test_transcript_is_carried_when_the_spec_accepts_it(self):
- captured = []
- client = self._client([self._alloc_http_error(), self._WAV], captured)
- with patch.object(audiocpp_client, "_family_spec",
- return_value={"options": {"request": [
- {"name": "reference_text"}]}}), \
- patch.object(AudioCppTTSClient, "_voice_transcript",
- return_value="The spoken reference text."):
- client._request_wav("Hello.")
- self.assertEqual(captured[1]["options"]["reference_text"],
- "The spoken reference text.")
-
- def test_transcript_is_omitted_for_option_validating_families(self):
- captured = []
- client = self._client([self._alloc_http_error(), self._WAV], captured,
- family="dramabox")
- with patch.object(audiocpp_client, "_family_spec",
- return_value={"options": {"request": [
- {"name": "seed"}]}}), \
- patch.object(AudioCppTTSClient, "_voice_transcript",
- return_value="The spoken reference text."):
- client._request_wav("Hello.")
- self.assertNotIn("reference_text", captured[1].get("options", {}))
+class ErrorBodyClassificationTests(unittest.TestCase):
+ """Deterministic-error detection runs on the FULL HTTP error body."""
+
+ def test_fragment_beyond_200_chars_is_still_classified(self):
+ # The deterministic fragment sits deep in a long server message;
+ # truncating before matching would misclassify it as retryable.
+ filler = "x" * 300
+ body = json.dumps({"error": {"message":
+ f"{filler} model contract spec not found for family"}})
+ error = audiocpp_request_error(500, body)
+ self.assertIsInstance(error, NonRetryableTTSError)
+ self.assertIn("not retryable", str(error))
+
+ def test_quoted_message_is_truncated_for_display(self):
+ body = json.dumps({"error": {"message": "y" * 1000}})
+ error = audiocpp_request_error(500, body)
+ self.assertNotIn("y" * 300, str(error))
+ self.assertLess(len(str(error)), 1000)
+
+ def test_http_error_body_helper_reads_whole_body(self):
+ body = b"z" * 500
+ exc = urllib.error.HTTPError("http://x", 500, "ISE", {},
+ io.BytesIO(body))
+ self.assertEqual(audiocpp_client._http_error_body(exc), body.decode())
+
+
+class VoiceForRunTests(unittest.TestCase):
+ """audiocpp_voice_for_run: the "All"-run per-model voice resolution."""
+
+ def test_design_takes_no_voice(self):
+ self.assertIsNone(audiocpp_client.audiocpp_voice_for_run(
+ "voxcpm2", "vdes", "Vox", "narrator", ["narrator"]))
+
+ def test_speaker_entry_takes_the_speaker_pick_or_first_speaker(self):
+ model = "Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF"
+ resolve = audiocpp_client.audiocpp_voice_for_run
+ self.assertEqual(resolve("qwen3_tts", "tts", model, "Vivian", []),
+ "Vivian")
+ self.assertEqual(resolve("qwen3_tts", "tts", model, "narrator", []),
+ QWEN3_TTS_SPEAKERS[0])
+
+ def test_clone_entry_takes_the_preset_pick_or_first_server_voice(self):
+ self.assertEqual(audiocpp_client.audiocpp_voice_for_run(
+ "higgs_audio_tts", "tts", "higgs", "narrator",
+ ["narrator", "other"]), "narrator")
+ self.assertEqual(audiocpp_client.audiocpp_voice_for_run(
+ "higgs_audio_tts", "tts", "higgs", "unknown", ["first", "x"]),
+ "first")
+ self.assertIsNone(audiocpp_client.audiocpp_voice_for_run(
+ "higgs_audio_tts", "tts", "higgs", "unknown", []))
+
+ def test_pure_tts_family_takes_no_voice(self):
+ self.assertIsNone(audiocpp_client.audiocpp_voice_for_run(
+ "supertonic", "tts", "supertonic", "narrator", ["narrator"]))
diff --git a/app/tests/test_viewkit.py b/app/tests/test_viewkit.py
index efebae6..842ded8 100644
--- a/app/tests/test_viewkit.py
+++ b/app/tests/test_viewkit.py
@@ -21,9 +21,15 @@ class _SyncThread:
def __init__(self, target, daemon=None):
self._target = target
+ self._alive = False
def start(self):
+ self._alive = True
self._target()
+ self._alive = False
+
+ def is_alive(self):
+ return self._alive
def join(self, timeout=None):
pass
@@ -121,6 +127,9 @@ class DefaultPromptCancelTests(unittest.TestCase):
joined = threading.Event()
class Worker:
+ def is_alive(self):
+ return True
+
def join(self, timeout=None):
joined.set()
generated by cgit v1.2.3 (git 2.25.1) at 2026-09-08 09:52:41 +0000