Fix tests for Windows

On Windows there are several different behaviors around subprocess.

- It cannot pickle locally defined functions/classes.
  (I thought this was also the case for Linux/macOS, but seems not)
- Module-level global variables do not inherit their values from
  the parent process.
- Even for simple async pipeline with single concurrency, the
  output order can be changed.

- In subprocess call, quoting is different.
- When writing to a NamedTemporaryFile, the file must be closed
  before writing.

Author: mthrok

tests/spdl_unittest/cuda/buffer_transfer_test.py

@@ -23,7 +23,7 @@
 DEFAULT_CUDA = 0
 
 CMDS = {
-    "audio": f"{FFMPEG_CLI} -hide_banner -y -f lavfi -i 'sine=frequency=1000:sample_rate=48000:duration=3' -c:a pcm_s16le sample.wav",
+    "audio": f"{FFMPEG_CLI} -hide_banner -y -f lavfi -i \"sine=frequency=1000:sample_rate=48000:duration=3\" -c:a pcm_s16le sample.wav",
     "video": f"{FFMPEG_CLI} -hide_banner -y -f lavfi -i testsrc -frames:v 1000 sample.mp4",
     "image": f"{FFMPEG_CLI} -hide_banner -y -f lavfi -i color=0x000000,format=gray -frames:v 1 sample.png",
 }

tests/spdl_unittest/dataloader/iterator_test.py

@@ -611,55 +611,59 @@ def done():
     assert result.status == _Status.ITERATION_FINISHED
 
 
-def test_terate_in_subprocess_initializer_failure():
-    def src_fn() -> Iterable[int]:
-        return SourceIterable(10)
+def _src1() -> Iterable[int]:
+    return SourceIterable(10)
 
-    def fail() -> None:
-        raise ValueError("Failed!")
 
+def _init1() -> None:
+    raise ValueError("Failed!")
+
+
+def test_iterate_in_subprocess_initializer_failure():
     with pytest.raises(RuntimeError, match=r"Initializer failed"):
-        iterate_in_subprocess(src_fn, buffer_size=1, timeout=3, initializer=fail)
+        iterate_in_subprocess(_src1, buffer_size=1, timeout=3, initializer=_init1)
 
 
-def test_iterate_in_subprocess_iterator_initialize_failure():
-    def src_fn() -> Iterator[int]:
+def _src2() -> Iterator[int]:
+    if True:
         raise ValueError("Failed!")
-        return SourceIterable(10)
+    return SourceIterable(10)
+
 
+def test_iterate_in_subprocess_iterator_initialize_failure():
     with pytest.raises(RuntimeError, match=r"Failed to create the iterable"):
-        iterate_in_subprocess(src_fn, buffer_size=1, timeout=3)
+        iterate_in_subprocess(_src2, buffer_size=1, timeout=3)
 
 
-def test_iterate_in_subprocess_generator_fail():
+def _src3() -> Iterable[int]:
     class SourceIterableFails(SourceIterable):
         def __iter__(self) -> Iterator[int]:
             raise ValueError("Failed!")
             yield from range(self.n)
 
-    def src_fn() -> Iterable[int]:
-        return SourceIterableFails(10)
+    return SourceIterableFails(10)
+
 
-    ite = iter(iterate_in_subprocess(src_fn, buffer_size=1, timeout=3))
+def test_iterate_in_subprocess_generator_fail():
+    ite = iter(iterate_in_subprocess(_src3, buffer_size=1, timeout=3))
 
     with pytest.raises(RuntimeError, match=r"Failed to fetch the next item"):
         next(ite)
 
 
-def test_iterate_in_subprocess_fail_after_n():
-    N = 10
-
+def _src4() -> Iterable[int]:
     class SourceIterableFails(SourceIterable):
         def __iter__(self) -> Iterator[int]:
             for v in range(self.n):
                 yield v
                 if v == 2:
                     raise ValueError("Failed!")
 
-    def src_fn() -> Iterable[int]:
-        return SourceIterableFails(N)
+    return SourceIterableFails(10)
+
 
-    ite = iter(iterate_in_subprocess(src_fn, buffer_size=1, timeout=3))
+def test_iterate_in_subprocess_fail_after_n():
+    ite = iter(iterate_in_subprocess(_src4, buffer_size=1, timeout=3))
     assert next(ite) == 0
     assert next(ite) == 1
     assert next(ite) == 2
@@ -668,13 +672,14 @@ def src_fn() -> Iterable[int]:
         next(ite)
 
 
+def _src5(N) -> Iterable[int]:
+    return SourceIterable(N)
+
+
 def test_iterate_in_subprocess_success():
     N = 3
 
-    def src_fn() -> Iterable[int]:
-        return SourceIterable(N)
-
-    hyp = list(iterate_in_subprocess(src_fn, buffer_size=-1, timeout=3))
+    hyp = list(iterate_in_subprocess(partial(_src5, N), buffer_size=-1, timeout=3))
     assert hyp == list(range(N))
 
 
@@ -684,13 +689,12 @@ def __iter__(self):
         yield 0
 
 
-def test_iterate_in_subprocess_timeout():
-    N = 3
+def _src6() -> Iterable[int]:
+    return SleepSourceIterable(3)
 
-    def src_fn() -> Iterable[int]:
-        return SleepSourceIterable(N)
 
-    iterable = iterate_in_subprocess(src_fn, buffer_size=-1, timeout=3)
+def test_iterate_in_subprocess_timeout():
+    iterable = iterate_in_subprocess(_src6, buffer_size=-1, timeout=3)
     iterator = iter(iterable)
     with pytest.raises(
         RuntimeError, match=r"The worker process did not produce any data for"

tests/spdl_unittest/dataloader/pipeline_test.py

@@ -11,6 +11,7 @@
 import platform
 import random
 import re
+import sys
 import threading
 import time
 from collections.abc import Iterator
@@ -19,6 +20,7 @@
 from functools import partial
 from multiprocessing import Process
 from typing import TypeVar
+from unittest import skipIf
 
 import pytest
 from spdl.pipeline import (
@@ -344,7 +346,7 @@ async def test(concurrency):
     # 1, 2, 3 and 4 in output_queue.
     remain, output = asyncio.run(test(4))
     assert remain == []
-    assert output == [1, 2, 3, 4]
+    assert set(output) == {1, 2, 3, 4}
 
 
 def test_async_pipe_concurrency_throughput():
@@ -377,7 +379,8 @@ async def test(concurrency):
 
         result = _flush_aqueue(output_queue)
 
-        assert result == ref
+        assert set(result) == set(ref)
+        assert result[-1] == ref[-1] == _EOF
 
         return elapsed
 
@@ -1944,11 +1947,20 @@ def test_run_pipeline_in_subprocess_state():
     assert src.src.seed == 2
 
 
-def _validate_pipeline_id(val: int) -> Iterator[int]:
-    assert _build._PIPELINE_ID == val
-    yield 0
+class _validate_pipeline_id:
+    def __init__(self, val: int) -> None:
+        self.val = val
+
+    def __iter__(self) -> Iterator[int]:
+        if _build._PIPELINE_ID != self.val:
+            raise AssertionError(f"{_build._PIPELINE_ID=} != {self.val=}")
+        yield 0
 
 
+# TODO: Fix this.
+@skipIf(
+    sys.platform == "win32", "On Windows module-level global variable is not inherited."
+)
 def test_run_pipeline_in_subprocess_pipeline_id():
     """The pipeline construdted in a subprocess inherits the global ID from the main process"""
     # Set to a number that's not zero and something unlikely to happen during the testing

tests/spdl_unittest/fixture.py

@@ -50,15 +50,15 @@ class SrcInfo:
 
 def get_sample(cmd: str) -> SrcInfo:
     samples = get_samples(cmd)
-    assert len(samples) == 1
+    assert len(samples) == 1, f"There must be one sample. Found: {len(samples)}"
     return samples[0]
 
 
 def get_samples(cmd: str) -> list[SrcInfo]:
     tmp_dir = TemporaryDirectory()
     tmp_path = Path(tmp_dir.name)
 
-    _run_in_tmpdir(cmd, tmp_path)
+    _run_in_tmpdir(cmd.strip(), tmp_path)
     return [SrcInfo(str(f), tmp_dir) for f in tmp_path.glob("**/*") if f.is_file()]
 
 

tests/spdl_unittest/io/async_test.py

@@ -52,7 +52,7 @@ def _test_decode(demux_fn, timestamps):
 
 def test_decode_audio_clips():
     """Can decode audio clips."""
-    cmd = f"{FFMPEG_CLI} -hide_banner -y -f lavfi -i 'sine=frequency=1000:sample_rate=48000:duration=3' -c:a pcm_s16le sample.wav"
+    cmd = f"{FFMPEG_CLI} -hide_banner -y -f lavfi -i sine=frequency=1000:sample_rate=48000:duration=3 -c:a pcm_s16le sample.wav"
     sample = get_sample(cmd)
 
     def _test():
@@ -71,7 +71,7 @@ def _test():
 
 def test_decode_audio_clips_num_frames():
     """Can decode audio clips with padding/dropping."""
-    cmd = f"{FFMPEG_CLI} -hide_banner -y -f lavfi -i 'sine=frequency=1000:sample_rate=16000:duration=1' -c:a pcm_s16le sample.wav"
+    cmd = f"{FFMPEG_CLI} -hide_banner -y -f lavfi -i sine=frequency=1000:sample_rate=16000:duration=1 -c:a pcm_s16le sample.wav"
     sample = get_sample(cmd)
 
     def _decode(src, num_frames=None):
@@ -284,7 +284,7 @@ def _test(src):
 
 def test_convert_audio():
     """convert_frames can convert AudioFrames to Buffer"""
-    cmd = f"{FFMPEG_CLI} -hide_banner -y -f lavfi -i 'sine=frequency=1000:sample_rate=48000:duration=3' -c:a pcm_s16le sample.wav"
+    cmd = f"{FFMPEG_CLI} -hide_banner -y -f lavfi -i sine=frequency=1000:sample_rate=48000:duration=3 -c:a pcm_s16le sample.wav"
     sample = get_sample(cmd)
 
     def _test(src):

tests/spdl_unittest/io/audio_decoding_test.py

@@ -32,8 +32,8 @@ def test_load_audio(sample_fmt):
     # fmt: off
     cmd = f"""
     {FFMPEG_CLI} -hide_banner -y \
-    -f lavfi -i 'sine=sample_rate=8000:frequency=305:duration=5' \
-    -f lavfi -i 'sine=sample_rate=8000:frequency=300:duration=5' \
+    -f lavfi -i sine=sample_rate=8000:frequency=305:duration=5 \
+    -f lavfi -i sine=sample_rate=8000:frequency=300:duration=5 \
     -filter_complex amerge  -c:a pcm_s16le sample.wav
     """
     # fmt: on
@@ -58,8 +58,8 @@ def test_batch_audio_conversion():
     # fmt: off
     cmd = f"""
     {FFMPEG_CLI} -hide_banner -y \
-    -f lavfi -i 'sine=sample_rate=8000:frequency=305:duration=5' \
-    -f lavfi -i 'sine=sample_rate=8000:frequency=300:duration=5' \
+    -f lavfi -i sine=sample_rate=8000:frequency=305:duration=5 \
+    -f lavfi -i sine=sample_rate=8000:frequency=300:duration=5 \
     -filter_complex amerge  -c:a pcm_s16le sample.wav
     """
     # fmt: on

tests/spdl_unittest/io/audio_encoding_test.py

@@ -44,6 +44,8 @@ def test_encode_audio_integer(sample_fmt):
     ref = np.random.randint(ii.min, ii.max, size=shape, dtype=dtype)
 
     with NamedTemporaryFile(suffix=".wav") as f:
+        f.close()  # for windows
+
         muxer = spdl.io.Muxer(f.name)
         encoder = muxer.add_encode_stream(
             config=spdl.io.audio_encode_config(
@@ -99,6 +101,8 @@ def test_encode_audio_float(sample_fmt):
     ref = np.random.rand(*shape).astype(dtype=dtype)
 
     with NamedTemporaryFile(suffix=".wav") as f:
+        f.close()  # for windows
+
         muxer = spdl.io.Muxer(f.name)
         encoder = muxer.add_encode_stream(
             config=spdl.io.audio_encode_config(
@@ -154,6 +158,8 @@ def test_encode_audio_integer_planar(sample_fmt):
     ref = np.random.randint(ii.min, ii.max, size=shape, dtype=dtype)
 
     with NamedTemporaryFile(suffix=".nut") as f:
+        f.close()  # windows
+
         muxer = spdl.io.Muxer(f.name)
         encoder = muxer.add_encode_stream(
             config=spdl.io.audio_encode_config(
@@ -219,13 +225,19 @@ def test_encode_audio_smoke_test(ext, sample_fmt):
         ref = np.random.random(shape).astype(dtype)
 
     with NamedTemporaryFile(suffix=ext) as f:
+        f.close()  # for windows
+
         muxer = spdl.io.Muxer(f.name)
         encoder = muxer.add_encode_stream(
             config=spdl.io.audio_encode_config(
                 num_channels=num_channels,
                 sample_fmt=sample_fmt,
                 sample_rate=sample_rate,
             ),
+            # on Windows, the default might be mp3_mf, which
+            # does not support planar format.
+            # So we specify lame
+            encoder="libmp3lame" if ext == ".mp3" else None
         )
 
         frame_size = encoder.frame_size or 1024
@@ -259,8 +271,8 @@ def test_remux_audio():
     # fmt: off
     cmd = f"""
     {FFMPEG_CLI} -hide_banner -y \
-    -f lavfi -i 'sine=sample_rate=8000:frequency=305:duration=5' \
-    -f lavfi -i 'sine=sample_rate=8000:frequency=300:duration=5' \
+    -f lavfi -i sine=sample_rate=8000:frequency=305:duration=5 \
+    -f lavfi -i sine=sample_rate=8000:frequency=300:duration=5 \
     -filter_complex amerge  -c:a pcm_s16le sample.wav
     """
     # fmt: on
@@ -269,6 +281,8 @@ def test_remux_audio():
     demuxer = spdl.io.Demuxer(sample.path)
 
     with NamedTemporaryFile(suffix=".wav") as f:
+        f.close()  # for windows
+
         muxer = spdl.io.Muxer(f.name)
         muxer.add_remux_stream(demuxer.audio_codec)
 

tests/spdl_unittest/io/configs_test.py

@@ -14,7 +14,7 @@
 
 def test_demux_config_smoketest():
     """"""
-    cmd = f"{FFMPEG_CLI} -hide_banner -y -f lavfi -i 'sine=frequency=1000:sample_rate=48000:duration=3' -c:a pcm_s16le sample.wav"
+    cmd = f"{FFMPEG_CLI} -hide_banner -y -f lavfi -i sine=frequency=1000:sample_rate=48000:duration=3 -c:a pcm_s16le sample.wav"
     sample = get_sample(cmd)
 
     demux_config = spdl.io.demux_config()
@@ -32,7 +32,7 @@ def test_demux_config_smoketest():
 
 def test_demux_config_headless():
     """Providing demux_config allows to load headeless audio"""
-    cmd = f"{FFMPEG_CLI} -hide_banner -y -f lavfi -i 'sine=frequency=1000:sample_rate=48000:duration=3' -f s16le -c:a pcm_s16le sample.raw"
+    cmd = f"{FFMPEG_CLI} -hide_banner -y -f lavfi -i sine=frequency=1000:sample_rate=48000:duration=3 -f s16le -c:a pcm_s16le sample.raw"
     sample = get_sample(cmd)
 
     with pytest.raises(RuntimeError):

tests/spdl_unittest/io/demuxer_test.py

@@ -14,7 +14,7 @@
 def test_demuxer_query_codec():
     """Can fetch the codec properly."""
     cmd = (
-        f"{FFMPEG_CLI} -hide_banner -y -f lavfi -i testsrc -f lavfi -i sine -t 5  sample.mp4",
+        f"{FFMPEG_CLI} -hide_banner -y -f lavfi -i testsrc -f lavfi -i sine -t 5  sample.mp4"
     )
 
     sample = get_sample(cmd)
@@ -40,7 +40,7 @@ def test_demuxer_query_codec():
 def test_demuxer_query_stream_index():
     """Can fetch the stream index properly."""
     cmd = (
-        f"{FFMPEG_CLI} -hide_banner -y -f lavfi -i testsrc -f lavfi -i sine -t 5  sample.mp4",
+        f"{FFMPEG_CLI} -hide_banner -y -f lavfi -i testsrc -f lavfi -i sine -t 5  sample.mp4"
     )
 
     sample = get_sample(cmd)
@@ -50,7 +50,7 @@ def test_demuxer_query_stream_index():
     assert demuxer.audio_stream_index == 1
 
     cmd = (
-        f"{FFMPEG_CLI} -hide_banner -y -f lavfi -i sine -f lavfi -i testsrc -t 5 -map 0:a -map 1:v  sample.mp4",
+        f"{FFMPEG_CLI} -hide_banner -y -f lavfi -i sine -f lavfi -i testsrc -t 5 -map 0:a -map 1:v  sample.mp4"
     )
 
     sample = get_sample(cmd)

tests/spdl_unittest/io/encoding_test.py

@@ -26,6 +26,7 @@ def test_encode_image_parity_simple(pix_fmt, torch_tensor):
     ref = np.random.randint(256, size=shape, dtype=np.uint8)
 
     with NamedTemporaryFile(suffix=".png") as f:
+        f.close()  # for windows
         spdl.io.save_image(
             f.name,
             torch.from_numpy(ref) if torch_tensor else ref,
@@ -44,6 +45,7 @@ def test_encode_image_parity_png_gray16be():
         ref = ref.byteswap()
 
     with NamedTemporaryFile(suffix=".png") as f:
+        f.close()  # for windows
         spdl.io.save_image(
             f.name,
             ref,
@@ -62,6 +64,7 @@ def test_encode_image_parity_png_gray16be():
 def _test_rejects(pix_fmt, dtype):
     data = np.ones((32, 64), dtype=dtype)
     with NamedTemporaryFile(suffix=".png") as f:
+        f.close()  # for windows
         with pytest.raises(ValueError):
             spdl.io.save_image(
                 f.name,