In test I/O utility, restore the old stdin/stdout instead of the "true" I/O streams (#5049)

I got a little bit nerdsniped by the problems observed in #5027. In
short, my high-level diagnosis in
#5027 (comment) seems
to have been correct: other tests were suppressing the legitimate
failure of a flaky test.

I found the problem by running other tests before the problem test, like
this:

```
$ pytest -k 'test_nonexistant_db or test_delete_removes_item' test/test_ui.py
```

When running `test_nonexistant_db` alone, it fails. When running it like
this with another test that goes first, it passes. That's the problem.

However, `test_delete_removes_item` is just one example that works to
make this problem happen. It appeared that _any_ test in a class that
used our `_common.TestCase` base class had this power. I tracked down
the issue to our `DummyIO` utility, which was having an unintentional
effect even when it was never actually used.

Here's the solution. Instead of restoring `sys.stdin` to
`sys.__stdin__`, we now restore it to whatever it was before we
installed out dummy I/O hooks. This is relevant in pytest, for example,
which installs its *own* `sys.stdin`, which we were then clobbering.
This was leading to the suppression of test failures observed in #5021
and addressed in #5027.

The CI will fail for this PR because it now (correctly) exposes a
failing test. Hopefully by combining this with the fixes in the works in
#5027, we'll be back to a passing test suite. 😃 @Phil305, could
you perhaps help validate that hypothesis?

Edit: @snejus:

I've now consolidated test I/O handling by removing the legacy
`control_stdin`/`capture_stdout` context managers and the custom
`DummyOut` stream, replacing them with a pytest-driven `io` fixture
that:
- provides controllable `stdin` via a lightweight `DummyIn`
- captures `stdout` via `capteesys`
- attaches a `DummyIO` helper to test classes as `self.io`

Author: snejus

beets/test/_common.py

@@ -14,10 +14,13 @@
 
 """Some common functionality for beets' test cases."""
 
+from __future__ import annotations
+
 import os
 import sys
 import unittest
 from contextlib import contextmanager
+from typing import TYPE_CHECKING
 
 import beets
 import beets.library
@@ -28,6 +31,9 @@
 from beets.ui import commands
 from beets.util import syspath
 
+if TYPE_CHECKING:
+    import pytest
+
 beetsplug.__path__ = [
     os.path.abspath(
         os.path.join(
@@ -118,85 +124,55 @@ def import_session(lib=None, loghandler=None, paths=[], query=[], cli=False):
 # Mock I/O.
 
 
-class InputError(Exception):
-    def __init__(self, output=None):
-        self.output = output
-
-    def __str__(self):
-        msg = "Attempt to read with no input provided."
-        if self.output is not None:
-            msg += f" Output: {self.output!r}"
-        return msg
-
-
-class DummyOut:
-    encoding = "utf-8"
-
-    def __init__(self):
-        self.buf = []
-
-    def write(self, s):
-        self.buf.append(s)
-
-    def get(self):
-        return "".join(self.buf)
-
-    def flush(self):
-        self.clear()
-
-    def clear(self):
-        self.buf = []
+class InputError(IOError):
+    def __str__(self) -> str:
+        return "Attempt to read with no input provided."
 
 
 class DummyIn:
     encoding = "utf-8"
 
-    def __init__(self, out=None):
-        self.buf = []
-        self.reads = 0
-        self.out = out
+    def __init__(self) -> None:
+        self.buf: list[str] = []
 
-    def add(self, s):
+    def add(self, s: str) -> None:
         self.buf.append(f"{s}\n")
 
-    def close(self):
+    def close(self) -> None:
         pass
 
-    def readline(self):
+    def readline(self) -> str:
         if not self.buf:
-            if self.out:
-                raise InputError(self.out.get())
-            else:
-                raise InputError()
-        self.reads += 1
+            raise InputError
+
         return self.buf.pop(0)
 
 
 class DummyIO:
-    """Mocks input and output streams for testing UI code."""
-
-    def __init__(self):
-        self.stdout = DummyOut()
-        self.stdin = DummyIn(self.stdout)
-
-    def addinput(self, s):
-        self.stdin.add(s)
-
-    def getoutput(self):
-        res = self.stdout.get()
-        self.stdout.clear()
-        return res
-
-    def readcount(self):
-        return self.stdin.reads
-
-    def install(self):
-        sys.stdin = self.stdin
-        sys.stdout = self.stdout
-
-    def restore(self):
-        sys.stdin = sys.__stdin__
-        sys.stdout = sys.__stdout__
+    """Test helper that manages standard input and output."""
+
+    def __init__(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        capteesys: pytest.CaptureFixture[str],
+    ) -> None:
+        self._capteesys = capteesys
+        self.stdin = DummyIn()
+
+        monkeypatch.setattr("sys.stdin", self.stdin)
+
+    def addinput(self, text: str) -> None:
+        """Simulate user typing into stdin."""
+        self.stdin.add(text)
+
+    def getoutput(self) -> str:
+        """Get the standard output captured so far.
+
+        Note: it clears the internal buffer, so subsequent calls will only
+        return *new* output.
+        """
+        # Using capteesys allows you to see output in the console if the test fails
+        return self._capteesys.readouterr().out
 
 
 # Utility.

beets/test/helper.py

@@ -15,9 +15,6 @@
 """This module includes various helpers that provide fixtures, capture
 information or mock the environment.
 
-- The `control_stdin` and `capture_stdout` context managers allow one to
-  interact with the user interface.
-
 - `has_program` checks the presence of a command on the system.
 
 - The `ImportSessionFixture` allows one to run importer code while
@@ -38,12 +35,12 @@
 from dataclasses import dataclass
 from enum import Enum
 from functools import cached_property
-from io import StringIO
 from pathlib import Path
 from tempfile import gettempdir, mkdtemp, mkstemp
 from typing import Any, ClassVar
 from unittest.mock import patch
 
+import pytest
 import responses
 from mediafile import Image, MediaFile
 
@@ -83,41 +80,6 @@ def capture_log(logger="beets"):
         log.removeHandler(capture)
 
 
-@contextmanager
-def control_stdin(input=None):
-    """Sends ``input`` to stdin.
-
-    >>> with control_stdin('yes'):
-    ...     input()
-    'yes'
-    """
-    org = sys.stdin
-    sys.stdin = StringIO(input)
-    try:
-        yield sys.stdin
-    finally:
-        sys.stdin = org
-
-
-@contextmanager
-def capture_stdout():
-    """Save stdout in a StringIO.
-
-    >>> with capture_stdout() as output:
-    ...     print('spam')
-    ...
-    >>> output.getvalue()
-    'spam'
-    """
-    org = sys.stdout
-    sys.stdout = capture = StringIO()
-    try:
-        yield sys.stdout
-    finally:
-        sys.stdout = org
-        print(capture.getvalue())
-
-
 def has_program(cmd, args=["--version"]):
     """Returns `True` if `cmd` can be executed."""
     full_cmd = [cmd, *args]
@@ -163,21 +125,31 @@ def config(self) -> beets.IncludeLazyConfig:
 )
 
 
-class IOMixin:
-    @cached_property
-    def io(self) -> _common.DummyIO:
-        return _common.DummyIO()
+class RunMixin:
+    def run_command(self, *args, **kwargs):
+        """Run a beets command with an arbitrary amount of arguments. The
+        Library` defaults to `self.lib`, but can be overridden with
+        the keyword argument `lib`.
+        """
+        sys.argv = ["beet"]  # avoid leakage from test suite args
+        lib = None
+        if hasattr(self, "lib"):
+            lib = self.lib
+        lib = kwargs.get("lib", lib)
+        beets.ui._raw_main(list(args), lib)
 
-    def setUp(self):
-        super().setUp()
-        self.io.install()
 
-    def tearDown(self):
-        super().tearDown()
-        self.io.restore()
+@pytest.mark.usefixtures("io")
+class IOMixin(RunMixin):
+    io: _common.DummyIO
+
+    def run_with_output(self, *args):
+        self.io.getoutput()
+        self.run_command(*args)
+        return self.io.getoutput()
 
 
-class TestHelper(ConfigMixin):
+class TestHelper(RunMixin, ConfigMixin):
     """Helper mixin for high-level cli and plugin tests.
 
     This mixin provides methods to isolate beets' global state provide
@@ -392,25 +364,6 @@ def create_mediafile_fixture(self, ext="mp3", images=[], target_dir=None):
 
         return path
 
-    # Running beets commands
-
-    def run_command(self, *args, **kwargs):
-        """Run a beets command with an arbitrary amount of arguments. The
-        Library` defaults to `self.lib`, but can be overridden with
-        the keyword argument `lib`.
-        """
-        sys.argv = ["beet"]  # avoid leakage from test suite args
-        lib = None
-        if hasattr(self, "lib"):
-            lib = self.lib
-        lib = kwargs.get("lib", lib)
-        beets.ui._raw_main(list(args), lib)
-
-    def run_with_output(self, *args):
-        with capture_stdout() as out:
-            self.run_command(*args)
-        return out.getvalue()
-
     # Safe file operations
 
     def create_temp_dir(self, **kwargs) -> str:
@@ -758,10 +711,7 @@ def _add_choice_input(self):
 class TerminalImportMixin(IOMixin, ImportHelper):
     """Provides_a terminal importer for the import session."""
 
-    io: _common.DummyIO
-
     def _get_import_session(self, import_dir: bytes) -> importer.ImportSession:
-        self.io.install()
         return TerminalImportSessionFixture(
             self.lib,
             loghandler=None,

test/conftest.py

@@ -5,6 +5,7 @@
 
 from beets.autotag.distance import Distance
 from beets.dbcore.query import Query
+from beets.test._common import DummyIO
 from beets.test.helper import ConfigMixin
 from beets.util import cached_classproperty
 
@@ -60,3 +61,24 @@ def clear_cached_classproperty():
 def config():
     """Provide a fresh beets configuration for a module, when requested."""
     return ConfigMixin().config
+
+
+@pytest.fixture
+def io(
+    request: pytest.FixtureRequest,
+    monkeypatch: pytest.MonkeyPatch,
+    capteesys: pytest.CaptureFixture[str],
+) -> DummyIO:
+    """Fixture for tests that need controllable stdin and captured stdout.
+
+    This fixture builds a per-test ``DummyIO`` helper and exposes it to the
+    test. When used on a test class, it attaches the helper as ``self.io``
+    attribute to make it available to all test methods, including
+    ``unittest.TestCase``-based ones.
+    """
+    io = DummyIO(monkeypatch, capteesys)
+
+    if request.instance:
+        request.instance.io = io
+
+    return io

test/plugins/test_bareasc.py

@@ -4,10 +4,10 @@
 """Tests for the 'bareasc' plugin."""
 
 from beets import logging
-from beets.test.helper import PluginTestCase, capture_stdout
+from beets.test.helper import IOMixin, PluginTestCase
 
 
-class BareascPluginTest(PluginTestCase):
+class BareascPluginTest(IOMixin, PluginTestCase):
     """Test bare ASCII query matching."""
 
     plugin = "bareasc"
@@ -65,16 +65,12 @@ def test_bareasc_search(self):
 
     def test_bareasc_list_output(self):
         """Bare-ASCII version of list command - check output."""
-        with capture_stdout() as output:
-            self.run_command("bareasc", "with accents")
+        self.run_command("bareasc", "with accents")
 
-        assert "Antonin Dvorak" in output.getvalue()
+        assert "Antonin Dvorak" in self.io.getoutput()
 
     def test_bareasc_format_output(self):
         """Bare-ASCII version of list -f command - check output."""
-        with capture_stdout() as output:
-            self.run_command(
-                "bareasc", "with accents", "-f", "$artist:: $title"
-            )
+        self.run_command("bareasc", "with accents", "-f", "$artist:: $title")
 
-        assert "Antonin Dvorak:: with accents\n" == output.getvalue()
+        assert "Antonin Dvorak:: with accents\n" == self.io.getoutput()

test/plugins/test_bpd.py

@@ -32,7 +32,7 @@
 from beets.test.helper import PluginTestCase
 from beets.util import bluelet
 
-bpd = pytest.importorskip("beetsplug.bpd")
+bpd = pytest.importorskip("beetsplug.bpd", exc_type=ImportError)
 
 
 class CommandParseTest(unittest.TestCase):