refactor (untested)

tests/test_batch.py

@@ -0,0 +1,109 @@
+"""Deeper tests for ctrl/batch.sh: recursive + space-safe discovery, audio
+inclusion, structure mirroring, flag forwarding, failure resilience, and the
+stdin-isolation fix (a greedy child must not stall the batch)."""
+import os
+import subprocess
+import tempfile
+import unittest
+from pathlib import Path
+
+from tests import REPO_ROOT
+
+BATCH = REPO_ROOT / "ctrl" / "batch.sh"
+
+
+def run_batch(args, env_extra=None):
+    env = dict(os.environ)
+    if env_extra:
+        env.update(env_extra)
+    return subprocess.run(["bash", str(BATCH), *args], cwd=REPO_ROOT, env=env,
+                          capture_output=True, text=True)
+
+
+def stub(d, body):
+    p = d / "pystub"
+    p.write_text("#!/bin/bash\n" + body + "\n")
+    p.chmod(0o755)
+    return str(p)
+
+
+class TestBatchDry(unittest.TestCase):
+    def test_recursive_spaces_audio_and_mirror(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp = Path(tmp)
+            ind = tmp / "in"
+            (ind / "team a" / "sub one").mkdir(parents=True)
+            (ind / "team a" / "sub one" / "session 1.mkv").touch()
+            (ind / "clip.MP4").touch()      # case-insensitive
+            (ind / "radio.ogg").touch()     # audio
+            (ind / "notes.txt").touch()     # ignored
+            r = run_batch(["-i", str(ind), "-o", str(tmp / "out"), "-n"])
+            self.assertEqual(r.returncode, 0, r.stderr)
+            out = r.stdout
+            self.assertIn("session 1.mkv", out)
+            self.assertIn("clip.MP4", out)
+            self.assertIn("radio.ogg", out)
+            self.assertNotIn("notes.txt", out)
+            self.assertIn("Found 3 video", out)
+            self.assertIn(str(tmp / "out" / "team a" / "sub one"), out)
+
+
+class TestBatchRun(unittest.TestCase):
+    def test_forward_flags_and_mirror(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp = Path(tmp)
+            ind = tmp / "in" / "team a"
+            ind.mkdir(parents=True)
+            (ind / "session 1.mkv").touch()
+            r = run_batch(["-i", str(tmp / "in"), "-o", str(tmp / "out"),
+                           "--", "--embed-images", "--diarize"],
+                          env_extra={"PYTHON": "echo"})
+            self.assertEqual(r.returncode, 0, r.stderr)
+            line = next(l for l in r.stdout.splitlines() if "process_meeting.py" in l)
+            self.assertIn("--output-dir", line)
+            self.assertIn(str(tmp / "out" / "team a"), line)
+            self.assertIn("--embed-images", line)
+            self.assertIn("session 1.mkv", line)
+
+    def test_empty_forward_does_not_crash(self):
+        # set -u + empty FORWARD array guard.
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp = Path(tmp)
+            ind = tmp / "in"
+            ind.mkdir()
+            (ind / "x.mp4").touch()
+            r = run_batch(["-i", str(ind), "-o", str(tmp / "out")],
+                          env_extra={"PYTHON": "echo"})
+            self.assertEqual(r.returncode, 0, r.stderr)
+            self.assertIn("process_meeting.py", r.stdout)
+
+    def test_continues_past_failures(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp = Path(tmp)
+            ind = tmp / "in"
+            ind.mkdir()
+            (ind / "a.mp4").touch()
+            (ind / "b.mp4").touch()
+            r = run_batch(["-i", str(ind), "-o", str(tmp / "out")],
+                          env_extra={"PYTHON": stub(tmp, "exit 1")})
+            self.assertNotEqual(r.returncode, 0)          # nonzero when any failed
+            self.assertIn("0 ok, 2 failed", r.stdout)
+            self.assertIn("FAILED", r.stderr)
+
+    def test_stdin_isolation_processes_all_items(self):
+        # A child that drains stdin (like ffmpeg) must NOT eat the file list.
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp = Path(tmp)
+            ind = tmp / "in"
+            ind.mkdir()
+            for n in ("a.mp4", "b.mp4", "c.mp4"):
+                (ind / n).touch()
+            greedy = stub(tmp, 'shift; cat >/dev/null; echo "DID $1"')
+            r = run_batch(["-i", str(ind), "-o", str(tmp / "out")],
+                          env_extra={"PYTHON": greedy})
+            self.assertEqual(r.returncode, 0, r.stderr)
+            self.assertEqual(r.stdout.count("DID "), 3)
+
+
+if __name__ == "__main__":
+    unittest.main()