Make sure tests do not leave any temp files behind

beets/test/helper.py

@@ -29,6 +29,7 @@
 - The `TestHelper` class encapsulates various fixtures that can be set up.
 """
 
+from __future__ import annotations
 
 import os
 import os.path
@@ -39,7 +40,9 @@
 from enum import Enum
 from io import StringIO
 from tempfile import mkdtemp, mkstemp
+from typing import ClassVar
 
+import responses
 from mediafile import Image, MediaFile
 
 import beets
@@ -49,7 +52,12 @@
 from beets.library import Album, Item, Library
 from beets.test import _common
 from beets.ui.commands import TerminalImportSession
-from beets.util import MoveOperation, bytestring_path, syspath
+from beets.util import (
+    MoveOperation,
+    bytestring_path,
+    clean_module_tempdir,
+    syspath,
+)
 
 
 class LogCapture(logging.Handler):
@@ -397,18 +405,14 @@ def add_album_fixture(
         return self.lib.add_album(items)
 
     def create_mediafile_fixture(self, ext="mp3", images=[]):
-        """Copies a fixture mediafile with the extension to a temporary
-        location and returns the path.
-
-        It keeps track of the created locations and will delete the with
-        `remove_mediafile_fixtures()`
+        """Copy a fixture mediafile with the extension to `temp_dir`.
 
         `images` is a subset of 'png', 'jpg', and 'tiff'. For each
         specified extension a cover art image is added to the media
         file.
         """
         src = os.path.join(_common.RSRC, util.bytestring_path("full." + ext))
-        handle, path = mkstemp()
+        handle, path = mkstemp(dir=self.temp_dir)
         path = bytestring_path(path)
         os.close(handle)
         shutil.copyfile(syspath(src), syspath(path))
@@ -424,17 +428,8 @@ def create_mediafile_fixture(self, ext="mp3", images=[]):
             mediafile.images = imgs
             mediafile.save()
 
-        if not hasattr(self, "_mediafile_fixtures"):
-            self._mediafile_fixtures = []
-        self._mediafile_fixtures.append(path)
-
         return path
 
-    def remove_mediafile_fixtures(self):
-        if hasattr(self, "_mediafile_fixtures"):
-            for path in self._mediafile_fixtures:
-                os.remove(syspath(path))
-
     def _get_item_count(self):
         if not hasattr(self, "__item_count"):
             count = 0
@@ -925,3 +920,39 @@ def _make_album_match(self, artist, album, tracks, distance=0, missing=0):
             albumtype="soundtrack",
             data_source="match_source",
         )
+
+
+class FetchImageHelper:
+    """Helper mixin for mocking requests when fetching images
+    with remote art sources.
+    """
+
+    @responses.activate
+    def run(self, *args, **kwargs):
+        super().run(*args, **kwargs)
+
+    IMAGEHEADER = {
+        "image/jpeg": b"\x00" * 6 + b"JFIF",
+        "image/png": b"\211PNG\r\n\032\n",
+    }
+
+    def mock_response(self, url, content_type="image/jpeg", file_type=None):
+        if file_type is None:
+            file_type = content_type
+        responses.add(
+            responses.GET,
+            url,
+            content_type=content_type,
+            # imghdr reads 32 bytes
+            body=self.IMAGEHEADER.get(file_type, b"").ljust(32, b"\x00"),
+        )
+
+
+class CleanupModulesMixin:
+    modules: ClassVar[tuple[str, ...]]
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        """Remove files created by the plugin."""
+        for module in cls.modules:
+            clean_module_tempdir(module)

beets/util/__init__.py

@@ -13,6 +13,7 @@
 # included in all copies or substantial portions of the Software.
 
 """Miscellaneous utility functions."""
+from __future__ import annotations
 
 import errno
 import fnmatch
@@ -26,9 +27,11 @@
 import tempfile
 import traceback
 from collections import Counter, namedtuple
+from contextlib import suppress
 from enum import Enum
 from logging import Logger
 from multiprocessing.pool import ThreadPool
+from pathlib import Path
 from typing import (
     Any,
     AnyStr,
@@ -58,6 +61,7 @@
 WINDOWS_MAGIC_PREFIX = "\\\\?\\"
 T = TypeVar("T")
 Bytes_or_String: TypeAlias = Union[str, bytes]
+PathLike = Union[str, bytes, Path]
 
 
 class HumanReadableException(Exception):
@@ -161,7 +165,7 @@
     """Provide the canonical form of the path suitable for storing in
     the database.
     """
     path = syspath(path, prefix=False)
     path = os.path.normpath(os.path.abspath(os.path.expanduser(path)))
     return bytestring_path(path)
 
@@ -175,7 +179,7 @@
 
     The argument should *not* be the result of a call to `syspath`.
     """
     out = []
     last_path = None
     while path:
         path = os.path.dirname(path)
@@ -192,17 +196,17 @@
 
 def sorted_walk(
     path: AnyStr,
     ignore: Sequence = (),
     ignore_hidden: bool = False,
     logger: Optional[Logger] = None,
 ) -> Generator[Tuple, None, None]:
     """Like `os.walk`, but yields things in case-insensitive sorted,
     breadth-first order.  Directory and file names matching any glob
     pattern in `ignore` are skipped. If `logger` is provided, then
     warning messages are logged there when a directory cannot be listed.
     """
     # Make sure the paths aren't Unicode strings.
     path = bytestring_path(path)
     ignore = [bytestring_path(i) for i in ignore]
 
     # Get all the directories and files at this level.
@@ -227,7 +231,7 @@
             if fnmatch.fnmatch(base, pat):
                 if logger:
                     logger.debug(
                         "ignoring {} due to ignore rule {}".format(base, pat)
                     )
                 skip = True
                 break
@@ -235,7 +239,7 @@
             continue
 
         # Add to output as either a file or a directory.
         cur = os.path.join(path, base)
         if (ignore_hidden and not hidden.is_hidden(cur)) or not ignore_hidden:
             if os.path.isdir(syspath(cur)):
                 dirs.append(base)
@@ -249,7 +253,7 @@
 
     # Recurse into directories.
     for base in dirs:
         cur = os.path.join(path, base)
         # yield from sorted_walk(...)
         yield from sorted_walk(cur, ignore, ignore_hidden, logger)
 
@@ -302,7 +306,7 @@
     emptiness. If root is not provided, then only path may be removed
     (i.e., no recursive removal).
     """
     path = normpath(path)
     if root is not None:
         root = normpath(root)
 
@@ -1076,3 +1080,46 @@
             self.cache[owner] = self.getter(owner)
 
         return self.cache[owner]
+
+
+def get_module_tempdir(module: str) -> Path:
+    """Return the temporary directory for the given module.
+
+    The directory is created within the `/tmp/beets/<module>` directory on
+    Linux (or the equivalent temporary directory on other systems).
+
+    Dots in the module name are replaced by underscores.
+    """
+    module = module.replace("beets.", "").replace(".", "_")
+    return Path(tempfile.gettempdir()) / "beets" / module
+
+
+def clean_module_tempdir(module: str) -> None:
+    """Clean the temporary directory for the given module."""
+    tempdir = get_module_tempdir(module)
+    shutil.rmtree(tempdir, ignore_errors=True)
+    with suppress(OSError):
+        # remove parent (/tmp/beets) directory if it is empty
+        tempdir.parent.rmdir()
+
+
+def get_temp_filename(
+    module: str,
+    prefix: str = "",
+    path: PathLike | None = None,
+    suffix: str = "",
+) -> bytes:
+    """Return temporary filename for the given module and prefix.
+
+    The filename starts with the given `prefix`.
+    If 'suffix' is given, it is used a the file extension.
+    If 'path' is given, we use the same suffix.
+    """
+    if not suffix and path:
+        suffix = Path(os.fsdecode(path)).suffix
+
+    tempdir = get_module_tempdir(module)
+    tempdir.mkdir(parents=True, exist_ok=True)
+
+    _, filename = tempfile.mkstemp(dir=tempdir, prefix=prefix, suffix=suffix)
+    return bytestring_path(filename)

beets/util/artresizer.py

@@ -22,11 +22,10 @@
 import re
 import subprocess
 from itertools import chain
-from tempfile import NamedTemporaryFile
 from urllib.parse import urlencode
 
 from beets import logging, util
-from beets.util import bytestring_path, displayable_path, syspath
+from beets.util import displayable_path, get_temp_filename, syspath
 
 PROXY_URL = "https://images.weserv.nl/"
 
@@ -48,15 +47,6 @@ def resize_url(url, maxwidth, quality=0):
     return "{}?{}".format(PROXY_URL, urlencode(params))
 
 
-def temp_file_for(path):
-    """Return an unused filename with the same extension as the
-    specified path.
-    """
-    ext = os.path.splitext(path)[1]
-    with NamedTemporaryFile(suffix=os.fsdecode(ext), delete=False) as f:
-        return bytestring_path(f.name)
-
-
 class LocalBackendNotAvailableError(Exception):
     pass
 
@@ -141,7 +131,9 @@ def resize(
         Use the ``magick`` program or ``convert`` on older versions. Return
         the output path of resized image.
         """
-        path_out = path_out or temp_file_for(path_in)
+        if not path_out:
+            path_out = get_temp_filename(__name__, "resize_IM_", path_in)
+
         log.debug(
             "artresizer: ImageMagick resizing {0} to {1}",
             displayable_path(path_in),
@@ -208,7 +200,8 @@ def get_size(self, path_in):
             return None
 
     def deinterlace(self, path_in, path_out=None):
-        path_out = path_out or temp_file_for(path_in)
+        if not path_out:
+            path_out = get_temp_filename(__name__, "deinterlace_IM_", path_in)
 
         cmd = self.convert_cmd + [
             syspath(path_in, prefix=False),
@@ -366,7 +359,9 @@ def resize(
         """Resize using Python Imaging Library (PIL).  Return the output path
         of resized image.
         """
-        path_out = path_out or temp_file_for(path_in)
+        if not path_out:
+            path_out = get_temp_filename(__name__, "resize_PIL_", path_in)
+
         from PIL import Image
 
         log.debug(
@@ -442,7 +437,9 @@ def get_size(self, path_in):
             return None
 
     def deinterlace(self, path_in, path_out=None):
-        path_out = path_out or temp_file_for(path_in)
+        if not path_out:
+            path_out = get_temp_filename(__name__, "deinterlace_PIL_", path_in)
+
         from PIL import Image
 
         try:

beetsplug/fetchart.py

@@ -19,14 +19,13 @@
 import re
 from collections import OrderedDict
 from contextlib import closing
-from tempfile import NamedTemporaryFile
 
 import confuse
 import requests
 from mediafile import image_mime_type
 
 from beets import config, importer, plugins, ui, util
-from beets.util import bytestring_path, sorted_walk, syspath
+from beets.util import bytestring_path, get_temp_filename, sorted_walk, syspath
 from beets.util.artresizer import ArtResizer
 
 try:
@@ -412,17 +411,17 @@ def fetch_image(self, candidate, plugin):
                         ext,
                     )
 
-                suffix = os.fsdecode(ext)
-                with NamedTemporaryFile(suffix=suffix, delete=False) as fh:
+                filename = get_temp_filename(__name__, suffix=ext.decode())
+                with open(filename, "wb") as fh:
                     # write the first already loaded part of the image
                     fh.write(header)
                     # download the remaining part of the image
                     for chunk in data:
                         fh.write(chunk)
                 self._log.debug(
-                    "downloaded art to: {0}", util.displayable_path(fh.name)
+                    "downloaded art to: {0}", util.displayable_path(filename)
                 )
-                candidate.path = util.bytestring_path(fh.name)
+                candidate.path = util.bytestring_path(filename)
                 return
 
         except (OSError, requests.RequestException, TypeError) as exc:

beetsplug/play.py

@@ -18,12 +18,12 @@
 import shlex
 import subprocess
 from os.path import relpath
-from tempfile import NamedTemporaryFile
 
 from beets import config, ui, util
 from beets.plugins import BeetsPlugin
 from beets.ui import Subcommand
 from beets.ui.commands import PromptChoice
+from beets.util import get_temp_filename
 
 # Indicate where arguments should be inserted into the command string.
 # If this is missing, they're placed at the end.
@@ -194,15 +194,15 @@ def _exceeds_threshold(
     def _create_tmp_playlist(self, paths_list):
         """Create a temporary .m3u file. Return the filename."""
         utf8_bom = config["play"]["bom"].get(bool)
-        m3u = NamedTemporaryFile("wb", suffix=".m3u", delete=False)
+        filename = get_temp_filename(__name__, suffix=".m3u")
+        with open(filename, "wb") as m3u:
+            if utf8_bom:
+                m3u.write(b"\xEF\xBB\xBF")
 
-        if utf8_bom:
-            m3u.write(b"\xEF\xBB\xBF")
+            for item in paths_list:
+                m3u.write(item + b"\n")
 
-        for item in paths_list:
-            m3u.write(item + b"\n")
-        m3u.close()
-        return m3u.name
+        return filename
 
     def before_choose_candidate_listener(self, session, task):
         """Append a "Play" choice to the interactive importer prompt."""

pyproject.toml

@@ -230,6 +230,21 @@ env.OPTS = """
 --cov-context=test
 """
 
+[tool.poe.tasks.check-temp-files]
+help = "Run each test module one by one and check for leftover temp files"
+shell = """
+setopt nullglob
+for file in test/**/*.py; do
+  print Temp files created by $file && poe test $file &>/dev/null
+  tempfiles=(/tmp/**/tmp* /tmp/beets/**/*)
+  if (( $#tempfiles )); then
+    print -l $'\t'$^tempfiles
+    rm -r --interactive=never $tempfiles &>/dev/null
+  fi
+done
+"""
+interpreter = "zsh"
+
 [tool.black]
 line-length = 80
 target-version = ["py38", "py39", "py310", "py311"]