Fix duplicate database change event send on Library.add (#5561)

## Description

Fixes #5560. Also a couple other incidental changes / improvements:
* Add `EventType` that holds the actual string literals used for event
sending. With type checking, this can prevent subtle bugs resulting from
misspelled event names.
* Fix `HiddenFileTest` by using `bytestring_path()`

## To Do

- [x] ~Documentation.~
- [x] Changelog.
- [x] Tests.

---------

Co-authored-by: J0J0 Todos <jojo@peek-a-boo.at>
Co-authored-by: J0J0 Todos <2733783+JOJ0@users.noreply.github.com>
This commit is contained in:
Ian McCowan 2025-05-30 06:41:29 -07:00 committed by GitHub
parent dd2f203090
commit 0f76312f31
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 66 additions and 5 deletions

33
beets/event_types.py Normal file
View file

@ -0,0 +1,33 @@
from typing import Literal
EventType = Literal[
"pluginload",
"import",
"album_imported",
"album_removed",
"item_copied",
"item_imported",
"before_item_moved",
"item_moved",
"item_linked",
"item_hardlinked",
"item_reflinked",
"item_removed",
"write",
"after_write",
"import_task_created",
"import_task_start",
"import_task_apply",
"import_task_before_choice",
"import_task_choice",
"import_task_files",
"library_opened",
"database_change",
"cli_exit",
"import_begin",
"trackinfo_received",
"albuminfo_received",
"before_choose_candidate",
"mb_track_extract",
"mb_album_extract",
]

View file

@ -369,8 +369,9 @@ class LibModel(dbcore.Model["Library"]):
plugins.send("database_change", lib=self._db, model=self)
def add(self, lib=None):
# super().add() calls self.store(), which sends `database_change`,
# so don't do it here
super().add(lib)
plugins.send("database_change", lib=self._db, model=self)
def __format__(self, spec):
if not spec:

View file

@ -39,6 +39,9 @@ import beets
from beets import logging
from beets.util.id_extractors import extract_release_id
if TYPE_CHECKING:
from beets.event_types import EventType
if sys.version_info >= (3, 10):
from typing import ParamSpec
else:
@ -292,7 +295,7 @@ class BeetsPlugin:
_raw_listeners: dict[str, list[Listener]] | None = None
listeners: dict[str, list[Listener]] | None = None
def register_listener(self, event: str, func: Listener) -> None:
def register_listener(self, event: "EventType", func: Listener):
"""Add a function as a listener for the specified event."""
wrapped_func = self._set_log_level_and_params(logging.WARNING, func)

View file

@ -31,6 +31,10 @@ Bug fixes:
:bug:`5788`
* tests: Fix library tests failing on Windows when run from outside ``D:/``.
:bug:`5802`
* Fix an issue where calling `Library.add` would cause the `database_change`
event to be sent twice, not once.
:bug:`5560`
* Fix ``HiddenFileTest`` by using ``bytestring_path()``.
For packagers:

View file

@ -22,7 +22,7 @@ import tempfile
import unittest
from beets import util
from beets.util import hidden
from beets.util import bytestring_path, hidden
class HiddenFileTest(unittest.TestCase):
@ -44,7 +44,7 @@ class HiddenFileTest(unittest.TestCase):
else:
raise e
assert hidden.is_hidden(f.name)
assert hidden.is_hidden(bytestring_path(f.name))
def test_windows_hidden(self):
if not sys.platform == "win32":

View file

@ -29,11 +29,12 @@ from mediafile import MediaFile, UnreadableFileError
import beets.dbcore.query
import beets.library
import beets.logging as blog
from beets import config, plugins, util
from beets.library import Album
from beets.test import _common
from beets.test._common import item
from beets.test.helper import BeetsTestCase, ItemInDBTestCase
from beets.test.helper import BeetsTestCase, ItemInDBTestCase, capture_log
from beets.util import as_string, bytestring_path, normpath, syspath
# Shortcut to path normalization.
@ -126,6 +127,25 @@ class AddTest(BeetsTestCase):
)
assert new_grouping == self.i.grouping
def test_library_add_one_database_change_event(self):
"""Test library.add emits only one database_change event."""
self.item = _common.item()
self.item.path = beets.util.normpath(
os.path.join(
self.temp_dir,
b"a",
b"b.mp3",
)
)
self.item.album = "a"
self.item.title = "b"
blog.getLogger("beets").set_global_level(blog.DEBUG)
with capture_log() as logs:
self.lib.add(self.item)
assert logs.count("Sending event: database_change") == 1
class RemoveTest(ItemInDBTestCase):
def test_remove_deletes_from_db(self):