test_pipeline_inbox.py

tests/test_pipeline_inbox.py

1. """The pipeline end to end with stand-in model steps, and the watch-folder run."""
3. import os
4. import threading
5. import time
7. import pytest
9. from transcription_pipeline import audio, cli, filing, inbox
10. from transcription_pipeline.pipeline import Options, process
12. from .helpers import fake_result, write_tone
15. def fake_transcriber(wav, source, opts):
16. assert wav.suffix == ".wav" and wav.exists()
17. return fake_result(speakers=False)
20. def fake_aligner(result, wav, opts):
21. return result
24. def fake_diarizer(result, wav, opts):
25. return fake_result(speakers=True)
28. def run(src, out, vault=None, **kw):
29. return process(src, out, Options(**kw), fake_transcriber, fake_aligner, fake_diarizer,
30. vault_inbox=vault)
33. def test_process_writes_every_format(tone, tmp_path):
34. res = run(tone, tmp_path / "out", formats=("txt", "md", "json", "srt", "vtt"))
35. assert set(res.outputs) == {"txt", "md", "json", "srt", "vtt"}
36. assert res.speakers == ["Speaker 1", "Speaker 2"] and res.sentences == 3
37. assert abs(res.duration - 1.5) < 0.01
38. assert res.outputs["txt"].read_text().startswith("[00:00:00] Speaker 1: Hi, thanks")
39. assert 'duration: "0:00:02"' in res.outputs["md"].read_text()
42. def test_no_diarize_skips_the_diarizer(tone, tmp_path):
43. res = run(tone, tmp_path / "out", diarize=False)
44. assert res.speakers == []
45. assert "Speaker" not in res.outputs["txt"].read_text()
48. def test_vault_copy_and_collisions(tone, tmp_path):
49. vault = tmp_path / "vault" / "Inbox"
50. a = run(tone, tmp_path / "out", vault=vault)
51. b = run(tone, tmp_path / "out", vault=vault)
52. assert a.outputs["vault"].name == "call.md" and b.outputs["vault"].name == "call (2).md"
53. assert b.outputs["txt"].name == "call (2).txt"
54. assert a.outputs["txt"].read_text() == b.outputs["txt"].read_text()
57. def test_video_input_is_decoded(tmp_path):
58. # Any container ffmpeg reads is fine: make an .m4a from the tone.
59. wav = write_tone(tmp_path / "t.wav")
60. m4a = tmp_path / "memo.m4a"
61. audio._run(["-i", str(wav), "-c:a", "aac", str(m4a)])
62. assert run(m4a, tmp_path / "out").duration > 1.0
65. def test_unique_path(tmp_path):
66. (tmp_path / "a.txt").write_text("x")
67. (tmp_path / "a (2).txt").write_text("x")
68. assert filing.unique_path(tmp_path, "a.txt").name == "a (3).txt"
71. def handler(layout, vault=None):
72. return lambda src: run(src, layout.transcripts, vault=vault)
75. def test_inbox_good_and_corrupt(tmp_path):
76. layout = inbox.Layout(tmp_path)
77. layout.ensure()
78. write_tone(layout.inbox / "good.wav")
79. (layout.inbox / "broken.mp3").write_bytes(b"this is not audio" * 100)
80. (layout.inbox / "notes.txt").write_text("ignored")
81. (layout.inbox / ".hidden.wav").write_bytes(b"x")
82. (layout.inbox / "big.mp3.part").write_bytes(b"x")
83. vault = tmp_path / "vault"
84. report = inbox.run(layout, handler(layout, vault), settle=0, log=lambda *_: None)
85. assert report == {"done": ["good.wav"], "failed": ["broken.mp3"], "skipped": None}
86. assert (layout.done / "good.wav").exists() and (layout.transcripts / "good.md").exists()
87. assert (vault / "good.md").exists()
88. assert (layout.failed / "broken.mp3").exists()
89. err = (layout.failed / "broken.error.txt").read_text()
90. assert err.startswith("AudioError: ffmpeg could not read this file")
91. left = sorted(p.name for p in layout.inbox.iterdir())
92. assert left == [".hidden.wav", "big.mp3.part", "notes.txt"]
93. assert not layout.lock.exists()
94. # A second run with nothing ready does nothing.
95. again = inbox.run(layout, handler(layout), settle=0, log=lambda *_: None)
96. assert again == {"done": [], "failed": [], "skipped": None}
99. def test_inbox_waits_for_a_growing_file(tmp_path):
100. layout = inbox.Layout(tmp_path)
101. layout.ensure()
102. growing = layout.inbox / "long.wav"
103. growing.write_bytes(b"RIFF")
104. stop = threading.Event()
106. def grow():
107. while not stop.is_set():
108. with open(growing, "ab") as f:
109. f.write(b"\0" * 1024)
110. time.sleep(0.02)
111. t = threading.Thread(target=grow)
112. t.start()
113. try:
114. report = inbox.run(layout, handler(layout), settle=0.3, log=lambda *_: None)
115. finally:
116. stop.set()
117. t.join()
118. assert report["done"] == [] and report["failed"] == [] and growing.exists()
121. def test_inbox_lock_held_by_live_process(tmp_path):
122. layout = inbox.Layout(tmp_path)
123. layout.ensure()
124. layout.lock.write_text(str(os.getpid()))
125. write_tone(layout.inbox / "a.wav")
126. report = inbox.run(layout, handler(layout), settle=0, log=lambda *_: None)
127. assert report["skipped"] == "locked" and (layout.inbox / "a.wav").exists()
130. def test_inbox_takes_over_a_stale_lock(tmp_path):
131. layout = inbox.Layout(tmp_path)
132. layout.ensure()
133. layout.lock.write_text("999999999")
134. write_tone(layout.inbox / "a.wav")
135. report = inbox.run(layout, handler(layout), settle=0, log=lambda *_: None)
136. assert report["done"] == ["a.wav"] and not layout.lock.exists()
139. def test_inbox_picks_up_files_that_arrive_mid_run(tmp_path):
140. layout = inbox.Layout(tmp_path)
141. layout.ensure()
142. write_tone(layout.inbox / "first.wav")
143. inner = handler(layout)
145. def h(src):
146. if src.name == "first.wav":
147. write_tone(layout.inbox / "second.wav")
148. return inner(src)
149. report = inbox.run(layout, h, settle=0, log=lambda *_: None)
150. assert report["done"] == ["first.wav", "second.wav"]
153. def test_cli_options():
154. args = cli.build_parser().parse_args(["x.m4a", "--speakers", "2", "--formats", "txt,.srt",
155. "--no-diarize", "--engine", "deepinfra"])
156. o = cli.options_from(args)
157. assert (o.min_speakers, o.max_speakers, o.formats, o.diarize, o.engine) == \
158. (2, 2, ("txt", "srt"), False, "deepinfra")
159. with pytest.raises(SystemExit):
160. cli.build_parser().parse_args(["x", "--formats", "docx"])
163. def test_cli_inbox_needs_a_root(monkeypatch, capsys):
164. monkeypatch.delenv("TP_ROOT", raising=False)
165. assert cli.main(["inbox", "--no-diarize"]) == 2
166. assert "TP_ROOT" in capsys.readouterr().err
169. def test_cli_missing_file(capsys):
170. assert cli.main(["/nonexistent/x.m4a", "--no-diarize"]) == 1