110 lines
4.2 KiB
Python
110 lines
4.2 KiB
Python
"""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()
|