Improve output tests (#285)

corrscope/corrscope.py

@@ -323,9 +323,6 @@ class CorrScope:
                             end_frame = frame + 1
                             break
 
-            if self.raise_on_teardown:
-                raise self.raise_on_teardown
-
         if PRINT_TIMESTAMP:
             # noinspection PyUnboundLocalVariable
             dtime_sec = time.perf_counter() - begin
@@ -338,5 +335,3 @@ class CorrScope:
                 msec_per_frame = float("inf")
 
             print(f"{frame_per_sec:.1f} FPS, {msec_per_frame:.2f} ms/frame")
-
-    raise_on_teardown: Optional[Exception] = None

setup.cfg

@@ -1,7 +1,11 @@
 [tool:pytest]
 testpaths = tests
-xfail_strict=true
 addopts = --tb=native
+xfail_strict=true
+
+;By default, pytest searches for tests within classes named Test*.
+;Turn that off.
+python_classes =
 
 [coverage:run]
 branch = True

tests/conftest.py

@@ -8,7 +8,7 @@ Integration tests found in:
 import os
 import subprocess
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional
 
 import pytest
 
@@ -23,18 +23,42 @@ os.chdir(Path(__file__).parent.parent)
 
 @pytest.fixture
 def Popen(mocker: "pytest_mock.MockFixture"):
+    """
+    - This fixture function is called Popen.
+        - pytest names our yield-value Popen too.
+    - The yielded Popen imitates `class Popen`, but is actually a MagicMock instance.
+        - We can't make Popen a class, because a test uses `Popen.call_args`.
+    - To "add class-methods" to Popen, I create a `class MockPopen`
+      and give its classmethods to Popen.
+    """
     real_Popen = subprocess.Popen
 
-    def popen_factory(*args, **kwargs):
-        popen = mocker.create_autospec(real_Popen)
+    class MockPopen(real_Popen):
+        exception_on_write: Optional[Exception] = None
 
-        popen.stdin = mocker.mock_open()(os.devnull, "wb")
-        popen.stdout = mocker.mock_open()(os.devnull, "rb")
-        assert popen.stdin != popen.stdout
+        @classmethod
+        def set_exception(cls, exc: Exception):
+            cls.exception_on_write = exc
 
-        popen.wait.return_value = 0
-        return popen
+        @classmethod
+        def __new__(cls, *args, **kwargs):
+            popen = mocker.create_autospec(real_Popen)
+
+            popen.stdin = mocker.mock_open()(os.devnull, "wb")
+            popen.stdout = mocker.mock_open()(os.devnull, "rb")
+            assert popen.stdin != popen.stdout
+
+            if cls.exception_on_write is not None:
+                popen.stdin.write.side_effect = cls.exception_on_write
+
+            popen.wait.return_value = 0
+            return popen
+
+    # Popen acts exactly the same as MockPopen
+    # when calling classmethods or creating instances.
+    # But Popen is a MagicMock, and captures Popen() call arguments.
 
     Popen = mocker.patch.object(subprocess, "Popen", autospec=True)
-    Popen.side_effect = popen_factory
+    Popen.set_exception = MockPopen.set_exception
+    Popen.side_effect = MockPopen.__new__
     yield Popen

tests/test_output.py

@@ -3,13 +3,12 @@
 - Integration tests (see conftest.py).
 """
 import errno
-import os
 import shutil
-import subprocess
 from fractions import Fraction
 from pathlib import Path
 from typing import TYPE_CHECKING
 
+import attr
 import pytest
 
 from corrscope.channel import ChannelConfig
@@ -24,12 +23,11 @@ from corrscope.outputs import (
 from corrscope.renderer import RendererConfig, Renderer
 from tests.test_renderer import RENDER_Y_ZEROS, WIDTH, HEIGHT
 
-
 if TYPE_CHECKING:
     import pytest_mock
-    from unittest.mock import MagicMock
 
 
+parametrize = pytest.mark.parametrize
 BYTES_PER_PIXEL = Renderer.bytes_per_pixel
 
 
@@ -42,26 +40,6 @@ if not shutil.which("ffmpeg"):
     )
 
 
-def exception_Popen(mocker: "pytest_mock.MockFixture", exc: Exception) -> "MagicMock":
-    """Mock Popen to raise an exception."""
-    real_Popen = subprocess.Popen
-
-    def popen_factory(*args, **kwargs):
-        popen = mocker.create_autospec(real_Popen)
-
-        popen.stdin = mocker.mock_open()(os.devnull, "wb")
-        popen.stdout = mocker.mock_open()(os.devnull, "rb")
-        assert popen.stdin != popen.stdout
-
-        popen.stdin.write.side_effect = exc
-        popen.wait.return_value = 0
-        return popen
-
-    Popen = mocker.patch.object(subprocess, "Popen", autospec=True)
-    Popen.side_effect = popen_factory
-    return Popen
-
-
 class DummyException(Exception):
     pass
 
@@ -183,16 +161,62 @@ def test_corr_terminate_ffplay(Popen, mocker: "pytest_mock.MockFixture"):
 
 
 # Integration: Calls CorrScope and FFplay.
-@pytest.mark.skip("Launches ffmpeg and ffplay processes, creating a ffplay window")
-def test_corr_terminate_works():
-    """ Ensure that ffmpeg/ffplay terminate quickly after Python exceptions, when
-    `popen.terminate()` is called. """
+
+
+@attr.dataclass(kw_only=True)
+class TestMode:
+    should_abort: bool = False
+    should_raise: bool = False
+
+
+@parametrize(
+    "test", [TestMode(should_abort=True), TestMode(should_raise=True)], ids=str
+)
+def test_corr_terminate_works(test):
+    """
+    Ensure that output exits quickly after output.terminate() is called.
+
+    What calls output.terminate() -> popen.terminate()?
+
+    - Cancelling a GUI render sets is_aborted()=True.
+    - corrscope may throw an exception.
+
+    Either way, ffmpeg should be terminated so it stops writing audio.
+    """
+
+    import sys
+    import subprocess
+    from corrscope.outputs import IOutputConfig, register_output, PipeOutput
+
+    class StayOpenOutputConfig(IOutputConfig):
+        pass
+
+    @register_output(StayOpenOutputConfig)
+    class StayOpenOutput(PipeOutput):
+        def __init__(self, corr_cfg: "Config", cfg: StayOpenOutputConfig):
+            super().__init__(corr_cfg, cfg)
+
+            sleep_process = subprocess.Popen(
+                [sys.executable, "-c", "import time; time.sleep(10)"],
+                stdin=subprocess.PIPE,
+            )
+            self.open(sleep_process)
+
+    def is_aborted() -> bool:
+        if test.should_raise:
+            raise DummyException
+        return test.should_abort
 
     cfg = sine440_config()
-    corr = CorrScope(cfg, Arguments(".", [FFplayOutputConfig()]))
-    corr.raise_on_teardown = DummyException
+    arg = Arguments(".", [StayOpenOutputConfig()], is_aborted=is_aborted)
+    corr = CorrScope(cfg, arg)
 
-    with pytest.raises(DummyException):
+    if test.should_raise:
+        with pytest.raises(DummyException):
+            # Raises `subprocess.TimeoutExpired` if popen.terminate() doesn't work.
+            corr.play()
+
+    else:
         # Raises `subprocess.TimeoutExpired` if popen.terminate() doesn't work.
         corr.play()
 
@@ -200,8 +224,9 @@ def test_corr_terminate_works():
 # Simulate user closing ffplay window.
 # Why OSError? See comment at PipeOutput.write_frame().
 # Calls FFplayOutput, mocks Popen.
+@pytest.mark.usefixtures("Popen")
 @pytest.mark.parametrize("errno_id", [errno.EPIPE, errno.EINVAL])
-def test_closing_ffplay_stops_main(mocker: "pytest_mock.MockFixture", errno_id):
+def test_closing_ffplay_stops_main(Popen, errno_id):
     """ Closing FFplay should make FFplayOutput.write_frame() return Stop
     to main loop. """
 
@@ -210,14 +235,12 @@ def test_closing_ffplay_stops_main(mocker: "pytest_mock.MockFixture", errno_id):
     if errno_id == errno.EPIPE:
         assert type(exc) == BrokenPipeError
 
-    # Yo Mock, I herd you like not working properly,
-    # so I put a test in your test so I can test your mocks while I test my code.
-    Popen = exception_Popen(mocker, exc)
-    assert Popen is subprocess.Popen
+    Popen.set_exception(exc)
     assert Popen.side_effect
 
     # Launch corrscope
     with FFplayOutputConfig()(CFG) as output:
+        # Writing to Popen instance raises exc.
         ret = output.write_frame(b"")
 
     # Ensure FFplayOutput catches OSError.