Merge pull request #5 from yukihiko-shinoda/hotfix-readme-example

Hotfix readme example

README.md

@@ -49,27 +49,37 @@ The package [`asynccpu`] is helpful to simple implement.
 Ex:
 
 ```python
+from asyncffmpeg.type_alias import StreamSpec
+from asyncffmpeg import FFmpegCoroutineFactory, StreamSpec
+
 import ffmpeg
 from asynccpu import ProcessTaskPoolExecutor
-from asyncffmpeg import FFmpegCoroutineFactory
 
-async def create_stream_spec_copy():
+
+async def create_stream_spec_copy() -> StreamSpec:
     stream = ffmpeg.input("input.mp4")
-    return ffmpeg.output(stream, "output.mp4", c="copy")
+    return ffmpeg.output(stream, "output1.mp4", c="copy")
+
 
-async def create_stream_spec_filter():
+async def create_stream_spec_filter() -> StreamSpec:
     stream = ffmpeg.input("input.mp4")
     stream = ffmpeg.filter(stream, "scale", 768, -1)
-    return ffmpeg.output(stream, "output.mp4")
+    return ffmpeg.output(stream, "output2.mp4")
+
+
+async def main() -> None:
+    ffmpeg_coroutine = FFmpegCoroutineFactory.create()
+
+    with ProcessTaskPoolExecutor(max_workers=3, cancel_tasks_when_shutdown=True) as executor:
+        awaitables = (
+            executor.create_process_task(ffmpeg_coroutine.execute, create_stream_spec)
+            for create_stream_spec in [create_stream_spec_copy, create_stream_spec_filter]
+        )
+        await asyncio.gather(*awaitables)
 
-ffmpeg_coroutine = FFmpegCoroutineFactory.create()
 
-with ProcessTaskPoolExecutor(max_workers=3, cancel_tasks_when_shutdown=True) as executor:
-    awaitables = {
-        executor.create_process_task(ffmpeg_coroutine, create_stream_spec)
-        for create_stream_spec in [create_stream_spec_copy, create_stream_spec_filter]
-    }
-    await asyncio.gather(*awaitables)
+if __name__ == "__main__":
+    asyncio.run(main())
 ```
 
 #### Why not [`asyncio`] but [`asynccpu`] ?

tests/conftest.py

@@ -1,5 +1,7 @@
 """Configuration of pytest"""
 import logging
+import os
+import shutil
 from contextlib import contextmanager
 from logging import handlers
 from multiprocessing import Queue
@@ -44,3 +46,24 @@ def ctx() -> Generator[None, None, None]:
             logger.handle(log_record)
 
     return ctx
+
+
+@pytest.fixture()
+def chdir(tmp_path: Path) -> Generator[Path, None, None]:
+    path_current = os.getcwd()
+    os.chdir(tmp_path)
+    yield tmp_path
+    os.chdir(path_current)
+
+
+@pytest.fixture()
+# Reason: To refer other fixture. pylint: disable=redefined-outer-name
+# Reason: This is fixture. pylint: disable=unused-argument
+def code_and_environment_example(
+    tmp_path: Path, resource_path_root: Path, path_file_input: Path, chdir: Path
+) -> Generator[Path, None, None]:
+    """Prepare files for test of example on README.md."""
+    path_sut = resource_path_root / "example_readme.py"
+    shutil.copy(path_sut, tmp_path)
+    shutil.copy(path_file_input, tmp_path / "input.mp4")
+    yield tmp_path / path_sut.name

tests/test_ffmpeg_coroutine.py

@@ -3,6 +3,7 @@
 import os
 import shutil
 import signal
+import subprocess
 import sys
 import time
 from logging import DEBUG
@@ -210,3 +211,9 @@ def terminate(cls, path_file_input: Path, path_file_output: Path) -> None:
     @staticmethod
     def report_raises_cencelled_error(path_file_input: Path, path_file_output: Path) -> None:
         asyncio.run(example_use_case(path_file_input, path_file_output))
+
+    @staticmethod
+    def test_example_readme(code_and_environment_example: Path) -> None:
+        subprocess.run([f"{sys.executable}", f"{code_and_environment_example}"], check=True)
+        for index in [1, 2]:
+            assert (code_and_environment_example.parent / f"output{index}.mp4").exists()

tests/testresources/example_readme.py

@@ -0,0 +1,35 @@
+import asyncio
+
+import ffmpeg
+
+# Reason: Following export method in __init__.py from Effective Python 2nd Edition item 85
+from asynccpu import ProcessTaskPoolExecutor  # type: ignore
+
+# Reason: Following export method in __init__.py from Effective Python 2nd Edition item 85
+from asyncffmpeg import FFmpegCoroutineFactory, StreamSpec  # type: ignore
+
+
+async def create_stream_spec_copy() -> StreamSpec:
+    stream = ffmpeg.input("input.mp4")
+    return ffmpeg.output(stream, "output1.mp4", c="copy")
+
+
+async def create_stream_spec_filter() -> StreamSpec:
+    stream = ffmpeg.input("input.mp4")
+    stream = ffmpeg.filter(stream, "scale", 768, -1)
+    return ffmpeg.output(stream, "output2.mp4")
+
+
+async def main() -> None:
+    ffmpeg_coroutine = FFmpegCoroutineFactory.create()
+
+    with ProcessTaskPoolExecutor(max_workers=3, cancel_tasks_when_shutdown=True) as executor:
+        awaitables = (
+            executor.create_process_task(ffmpeg_coroutine.execute, create_stream_spec)
+            for create_stream_spec in [create_stream_spec_copy, create_stream_spec_filter]
+        )
+        await asyncio.gather(*awaitables)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())