Allow specifying reencode's tmp dir

src/zimscraperlib/video/encoding.py

@@ -5,6 +5,7 @@
 from copy import deepcopy
 
 from zimscraperlib import logger
+from zimscraperlib.filesystem import path_from
 from zimscraperlib.logging import nicer_args_join
 
 
@@ -40,6 +41,7 @@ def reencode(
     *,
     delete_src: bool = False,
     failsafe: bool = True,
+    existing_tmp_path: pathlib.Path | None = None,
 ) -> tuple[bool, subprocess.CompletedProcess[str]]:
     """Runs ffmpeg with given ffmpeg_args
 
@@ -52,7 +54,8 @@ def reencode(
         failsafe - Run in failsafe mode
     """
 
-    with tempfile.TemporaryDirectory() as tmp_dir:
+    with path_from(existing_tmp_path or tempfile.TemporaryDirectory()) as tmp_dir:
+
         tmp_path = pathlib.Path(tmp_dir).joinpath(f"video.tmp{dst_path.suffix}")
         args = _build_ffmpeg_args(
             src_path=src_path,

tests/video/test_video.py

@@ -29,12 +29,19 @@ def copy_media_and_reencode(
     dest: str,
     ffmpeg_args: list[str],
     test_files: dict[str, pathlib.Path],
+    *,
+    use_temp_dir_for_temp_file: bool = False,
     **kwargs: Any,
 ):
     src_path = temp_dir.joinpath(src)
     dest_path = temp_dir.joinpath(dest)
     shutil.copy2(test_files[src_path.suffix[1:]], src_path)
-    return reencode(src_path, dest_path, ffmpeg_args, **kwargs)
+    if use_temp_dir_for_temp_file:
+        return reencode(
+            src_path, dest_path, ffmpeg_args, existing_tmp_path=temp_dir, **kwargs
+        )
+    else:
+        return reencode(src_path, dest_path, ffmpeg_args, **kwargs)
 
 
 def test_config_defaults():
@@ -392,6 +399,23 @@ def test_reencode_media(
         assert expected["codecs"] == converted_details["codecs"]
 
 
+@pytest.mark.slow
+def test_reencode_media_with_tmp_dir(test_files: dict[str, pathlib.Path]):
+    with tempfile.TemporaryDirectory() as t:
+        temp_dir = pathlib.Path(t)
+        copy_media_and_reencode(
+            temp_dir,
+            "video.mp4",
+            "video.webm",
+            VideoWebmLow().to_ffmpeg_args(),
+            test_files,
+            use_temp_dir_for_temp_file=True,
+        )
+        converted_details = get_media_info(temp_dir.joinpath("video.webm"))
+        assert converted_details["duration"] == 2
+        assert converted_details["codecs"] == ["vp9", "vorbis"]
+
+
 @pytest.mark.slow
 @pytest.mark.parametrize(
     "src,dest,ffmpeg_args,delete_src",