diff --git a/src/palace/manager/scripts/suppress.py b/src/palace/manager/scripts/suppress.py index 25acef596c..2ddb6957a5 100644 --- a/src/palace/manager/scripts/suppress.py +++ b/src/palace/manager/scripts/suppress.py @@ -3,7 +3,7 @@ from collections.abc import Sequence from datetime import datetime, timezone from enum import Enum, auto -from typing import cast +from typing import Any, NamedTuple, cast from sqlalchemy import select from sqlalchemy.orm import Session @@ -13,12 +13,25 @@ from palace.manager.scripts.base import Script, _normalize_cmd_args from palace.manager.sqlalchemy.model.identifier import Identifier from palace.manager.sqlalchemy.model.library import Library +from palace.manager.sqlalchemy.model.licensing import LicensePool +from palace.manager.sqlalchemy.model.work import Work class SuppressResult(Enum): NEWLY_SUPPRESSED = auto() ALREADY_SUPPRESSED = auto() NOT_FOUND = auto() + NOT_IN_LIBRARY = auto() + AMBIGUOUS = auto() + + +class SuppressOutcome(NamedTuple): + result: SuppressResult + # The title and work id of every work described by this outcome, + # formatted as " (work id: <id>)" and joined with "; " when + # there's more than one -- e.g. for AMBIGUOUS, or when + # --suppress-ambiguous affects more than one work at once. + description: str | None = None class SuppressWorkForLibraryScript(Script): @@ -26,6 +39,10 @@ class SuppressWorkForLibraryScript(Script): BY_DATABASE_ID = "Database ID" + def __init__(self, *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self._collection_ids: dict[int, list[int]] = {} + @classmethod def arg_parser(cls, _db: Session) -> argparse.ArgumentParser: parser = argparse.ArgumentParser() @@ -66,6 +83,14 @@ def arg_parser(cls, _db: Session) -> argparse.ArgumentParser: help="Report what would be suppressed without making any changes.", action="store_true", ) + parser.add_argument( + "--suppress-ambiguous", + help="If an identifier resolves to more than one distinct work " + "(e.g. the same title licensed through more than one of the " + "library's collections), suppress it in all of them instead of " + "skipping it.", + action="store_true", + ) return parser @classmethod @@ -142,34 +167,226 @@ def load_identifiers_from_file( raise PalaceValueError(f"CSV file not found: {file_path}") return identifiers + def _library_collection_ids(self, library: Library) -> list[int]: + """The ids of the collections a library is associated with. + + Cached for the life of the script: `Library.associated_collections` + issues a fresh SELECT on every access, and this is consulted once + per identifier, so a `--file` run would otherwise repeat the same + query for every row. + + Collections are scoped by association rather than by whether + they're currently active: `suppressed_for` is a durable flag, not + something that should depend on where today falls in a + collection's subscription window. + """ + if library.id not in self._collection_ids: + self._collection_ids[library.id] = [ + c.id for c in library.associated_collections if c.id is not None + ] + return self._collection_ids[library.id] + + def load_works(self, identifier: Identifier, library: Library) -> list[Work]: + """Find the Work(s) `library` carries for an identifier. + + Only works the library actually licenses are considered, so that a + title carried by two libraries through two different collections + (e.g. one via OverDrive, one via Bibliotheca) resolves cleanly for + each of them instead of looking ambiguous to both. + + Pools for this exact identifier in one of the library's collections + are exact matches, and win outright over equivalency. With + consistent data there is at most one Work among them, because + `LicensePool.calculate_work` forces every pool sharing an + identifier onto the same Work -- so suppressing it already covers + every collection that identifier appears in, a consortium's + OverDrive collection and a library's own Advantage collection + alike. Every matching work is nonetheless returned rather than the + first one found, so that inconsistent data (a state + `calculate_work` itself warns about and repairs, logging that the + pools have "more than one Work between them") meets the ambiguity + guard instead of being silently resolved to one of them. + + The legitimate way one identifier reaches several works is + different vendor identifiers sharing an ISBN, and that goes + through the equivalency path below rather than this one. + + A pool whose Work hasn't been calculated yet counts for nothing + here: suppression is a Work/Library relation, so a pool without a + Work has nothing that *can* be suppressed. Such a pool therefore + doesn't make this an exact match, and resolution falls through to + equivalency rather than refusing -- otherwise a title the library + demonstrably carries under an equivalent identifier would go + unsuppressed because some other pool of the same identifier was + mid-import. + + Failing an exact match, we look past LicensePools whose own + identifier matches, to also include LicensePools reachable through + identifier equivalency -- e.g. an ISBN a librarian has on hand is + often linked to a vendor's LicensePool via metadata equivalency + rather than being that LicensePool's own identifier. This uses the + same strict, high-confidence equivalency policy that + `Work.from_identifiers` applies by default everywhere else in the + codebase, so it won't walk into loosely-related works. + + Exact matches deliberately don't get combined with equivalency + matches: two pools of the same identifier are certainly the same + title, whereas an equivalency edge is only ever a vendor's + assertion, so widening an exact hit with equivalent works could + suppress an unrelated title. + """ + collection_ids = self._library_collection_ids(library) + if not collection_ids: + return [] + + direct_works = { + pool.work.id: pool.work + for pool in identifier.licensed_through + if pool.work is not None and pool.collection_id in collection_ids + } + if direct_works: + return [direct_works[work_id] for work_id in sorted(direct_works)] + + # The collection scope is an EXISTS rather than a filter on the + # joined pool, because the pool carrying the equivalent identifier + # and the pool in this library's collection may be different pools + # of the same Work. + query = ( + Work.from_identifiers(self._db, [identifier]) + .filter( + Work.license_pools.any(LicensePool.collection_id.in_(collection_ids)) + ) + .order_by(Work.id) + ) + return cast(list[Work], query.distinct().all()) + + @staticmethod + def _describe_works(works: list[Work]) -> str: + """Format a list of works for operator-facing output, e.g. for an + ambiguous match or when suppressing more than one work at once.""" + return "; ".join(f"{w.title or '[no title]'} (work id: {w.id})" for w in works) + + def _nothing_to_suppress( + self, identifier: Identifier, library: Library + ) -> SuppressOutcome: + """Report why an identifier resolved to none of the library's works. + + A title the library doesn't carry is a different problem from an + identifier that matches nothing at all -- the first means the + identifier is fine but belongs to someone else's collection, the + second usually means a typo -- so they get distinct results rather + than both landing in one "not found" bucket. + """ + elsewhere = ( + Work.from_identifiers(self._db, [identifier]).order_by(Work.id).first() + ) + if elsewhere is not None: + self.log.warning( + f"{identifier.type}/{identifier.identifier} resolves to a work " + f"that {library.short_name} does not carry in any of its collections." + ) + return SuppressOutcome( + SuppressResult.NOT_IN_LIBRARY, self._describe_works([elsewhere]) + ) + + self.log.warning(f"No work found for {identifier}") + return SuppressOutcome(SuppressResult.NOT_FOUND) + def suppress_work( self, library: Library, identifier: Identifier, dry_run: bool = False, - ) -> SuppressResult: - work = identifier.work - if not work: - self.log.warning(f"No work found for {identifier}") - return SuppressResult.NOT_FOUND + suppress_ambiguous: bool = False, + ) -> SuppressOutcome: + """Suppress the work(s) resolved from an identifier for a library. + + :param library: The library for which the resolved work should be suppressed. + :param identifier: The identifier used to resolve the work, either + directly (it owns a LicensePool in one of the library's + collections) or through identifier equivalency. + :param dry_run: If true, report the outcome without changing suppression. + :param suppress_ambiguous: If the identifier resolves to more than one + distinct work, suppress it for the library in every candidate + work instead of refusing to guess which one is meant. + :return: The result of the suppression attempt, and a description of + the work(s) it applies to, when any were resolved. + """ + works = self.load_works(identifier, library) + if not works: + return self._nothing_to_suppress(identifier, library) + + if len(works) == 1: + work = works[0] + description = self._describe_works(works) + + if library in work.suppressed_for: + return SuppressOutcome(SuppressResult.ALREADY_SUPPRESSED, description) + + if not dry_run: + # Suppression is scoped to exactly this one library. Resolving + # the work via identifier equivalency only changes *which work* + # we find -- it never changes *which libraries* it's suppressed + # for, since only the `library` argument passed in is ever + # appended to `suppressed_for`. + work.suppressed_for.append(library) + + self.log.info( + f"{'[DRY RUN] Would suppress' if dry_run else 'Suppressing'} " + f"{identifier.type}/{identifier.identifier} (work id: {work.id}) " + f"for {library.short_name}." + ) + return SuppressOutcome(SuppressResult.NEWLY_SUPPRESSED, description) + + # This identifier resolves to more than one of the library's works + # (see load_works) -- usually different vendor identifiers sharing + # an ISBN, each with its own Work. + if all(library in work.suppressed_for for work in works): + # Every candidate already reflects the desired state, so there's + # nothing to decide or change -- this isn't really ambiguous in + # practice, just a no-op re-run (e.g. after an earlier + # --suppress-ambiguous run already covered every candidate). + return SuppressOutcome( + SuppressResult.ALREADY_SUPPRESSED, self._describe_works(works) + ) - if library in work.suppressed_for: - return SuppressResult.ALREADY_SUPPRESSED + if not suppress_ambiguous: + # Without --suppress-ambiguous, refuse to guess which one(s) the + # operator meant. + self.log.warning( + f"{identifier.type}/{identifier.identifier} resolves to " + f"{len(works)} different works for {library.short_name}; " + "skipping rather than guessing which one to suppress. Pass " + "--suppress-ambiguous to suppress all of them for this library." + ) + return SuppressOutcome( + SuppressResult.AMBIGUOUS, self._describe_works(works) + ) - if not dry_run: - work.suppressed_for.append(library) + # At least one candidate isn't yet suppressed (the all-suppressed + # case was handled above), so this always changes at least one work. + changed_works: list[Work] = [] + for work in works: + if library in work.suppressed_for: + continue + if not dry_run: + work.suppressed_for.append(library) + changed_works.append(work) self.log.info( f"{'[DRY RUN] Would suppress' if dry_run else 'Suppressing'} " - f"{identifier.type}/{identifier.identifier} (work id: {work.id}) " - f"for {library.short_name}." + f"{identifier.type}/{identifier.identifier} in {len(changed_works)} " + f"of {len(works)} ambiguous work(s) for {library.short_name}." + ) + return SuppressOutcome( + SuppressResult.NEWLY_SUPPRESSED, self._describe_works(changed_works) ) - return SuppressResult.NEWLY_SUPPRESSED def do_run(self, cmd_args: list[str] | None = None) -> None: parsed = self.parse_command_line(self._db, cmd_args=cmd_args) library = self.load_library(parsed.library) dry_run: bool = parsed.dry_run + suppress_ambiguous: bool = parsed.suppress_ambiguous started_at = datetime.now(tz=timezone.utc) if parsed.file: @@ -189,15 +406,20 @@ def do_run(self, cmd_args: list[str] | None = None) -> None: ) pairs = unique_pairs - results: dict[tuple[str, str], SuppressResult] = {} + results: dict[tuple[str, str], SuppressOutcome] = {} try: for id_type, id_value in pairs: try: identifier = self.load_identifier(id_type, id_value) - result = self.suppress_work(library, identifier, dry_run=dry_run) + outcome = self.suppress_work( + library, + identifier, + dry_run=dry_run, + suppress_ambiguous=suppress_ambiguous, + ) except PalaceValueError: - result = SuppressResult.NOT_FOUND - results[(id_type, id_value)] = result + outcome = SuppressOutcome(SuppressResult.NOT_FOUND) + results[(id_type, id_value)] = outcome if not dry_run: self._db.commit() @@ -212,19 +434,29 @@ def do_run(self, cmd_args: list[str] | None = None) -> None: def _print_results( self, - results: dict[tuple[str, str], SuppressResult], + results: dict[tuple[str, str], SuppressOutcome], dry_run: bool, library: Library, started_at: datetime, duration_seconds: float, ) -> None: newly_suppressed = [ - k for k, v in results.items() if v == SuppressResult.NEWLY_SUPPRESSED + k for k, v in results.items() if v.result == SuppressResult.NEWLY_SUPPRESSED ] already_suppressed = [ - k for k, v in results.items() if v == SuppressResult.ALREADY_SUPPRESSED + k + for k, v in results.items() + if v.result == SuppressResult.ALREADY_SUPPRESSED + ] + not_found = [ + k for k, v in results.items() if v.result == SuppressResult.NOT_FOUND + ] + not_in_library = [ + k for k, v in results.items() if v.result == SuppressResult.NOT_IN_LIBRARY + ] + ambiguous = [ + k for k, v in results.items() if v.result == SuppressResult.AMBIGUOUS ] - not_found = [k for k, v in results.items() if v == SuppressResult.NOT_FOUND] prefix = "[DRY RUN] " if dry_run else "" suppress_label = "Would suppress" if dry_run else "Newly suppressed" @@ -236,6 +468,8 @@ def _print_results( (suppress_label + ":", len(newly_suppressed)), ("Already suppressed:", len(already_suppressed)), ("Not found:", len(not_found)), + ("Not in this library:", len(not_in_library)), + ("Ambiguous:", len(ambiguous)), ] col = max(len(label) for label, _ in summary_rows) print(f"\n{prefix}Suppression Results Summary:") @@ -249,7 +483,10 @@ def _print_results( ), SuppressResult.ALREADY_SUPPRESSED: "ALREADY SUPPRESSED", SuppressResult.NOT_FOUND: "NOT FOUND", + SuppressResult.NOT_IN_LIBRARY: "NOT IN THIS LIBRARY", + SuppressResult.AMBIGUOUS: "AMBIGUOUS", } - for (id_type, id_value), result in results.items(): - status = status_map[result] - print(f" [{status}] {id_type}/{id_value}") + for (id_type, id_value), outcome in results.items(): + status = status_map[outcome.result] + suffix = f" -- {outcome.description}" if outcome.description else "" + print(f" [{status}] {id_type}/{id_value}{suffix}") diff --git a/tests/manager/scripts/test_suppress.py b/tests/manager/scripts/test_suppress.py index 11894cd955..70c773c3f7 100644 --- a/tests/manager/scripts/test_suppress.py +++ b/tests/manager/scripts/test_suppress.py @@ -3,14 +3,68 @@ import re import textwrap from datetime import datetime, timezone +from pathlib import Path from unittest.mock import create_autospec, patch import pytest -from palace.manager.scripts.suppress import SuppressResult, SuppressWorkForLibraryScript +from palace.manager.scripts.suppress import ( + SuppressOutcome, + SuppressResult, + SuppressWorkForLibraryScript, +) +from palace.manager.sqlalchemy.model.datasource import DataSource +from palace.manager.sqlalchemy.model.identifier import Identifier +from palace.manager.sqlalchemy.model.library import Library +from palace.manager.sqlalchemy.model.work import Work from tests.fixtures.database import DatabaseTransactionFixture +def isbn_equivalent_to_works( + db: DatabaseTransactionFixture, library: Library, work_count: int = 2 +) -> tuple[Identifier, list[Work]]: + """An ISBN with no LicensePool of its own, linked by equivalency to + `work_count` distinct works licensed to `library`. + + This is the shape a librarian hits in practice: the ISBN they have in + hand isn't any pool's own identifier, and more than one work can hang + off it. + """ + collection = db.collection(library=library) + works = [ + db.work(with_license_pool=True, collection=collection) + for _ in range(work_count) + ] + isbn = db.identifier(identifier_type="ISBN") + source = DataSource.lookup(db.session, DataSource.OCLC) + for work in works: + isbn.equivalent_to(source, work.presentation_edition.primary_identifier, 1) + return isbn, works + + +def works_with_identifier_csv( + db: DatabaseTransactionFixture, + library: Library, + tmp_path: Path, + *, + repeat_first: bool = False, +) -> tuple[list[Work], str]: + """Two works licensed to `library`, plus the path to a CSV naming + their identifiers. `repeat_first` duplicates the first row.""" + collection = db.collection(library=library) + works = [db.work(with_license_pool=True, collection=collection) for _ in range(2)] + identifiers = [work.presentation_edition.primary_identifier for work in works] + if repeat_first: + identifiers.insert(1, identifiers[0]) + + csv_file = tmp_path / "ids.csv" + csv_file.write_text( + "identifier,identifier_type\n" + + "".join(f"{i.identifier},{i.type}\n" for i in identifiers) + ) + return works, str(csv_file) + + class TestSuppressWorkForLibraryScript: @pytest.mark.parametrize( "cmd_args", @@ -64,12 +118,34 @@ def test_parse_command_line_with_file(self, db: DatabaseTransactionFixture): assert parsed.file == "/tmp/ids.csv" assert parsed.identifier is None - def test_parse_command_line_dry_run(self, db: DatabaseTransactionFixture): + @pytest.mark.parametrize( + "extra_args,attribute,expected", + [ + pytest.param([], "dry_run", False, id="dry-run-default"), + pytest.param(["--dry-run"], "dry_run", True, id="dry-run-set"), + pytest.param( + [], "suppress_ambiguous", False, id="suppress-ambiguous-default" + ), + pytest.param( + ["--suppress-ambiguous"], + "suppress_ambiguous", + True, + id="suppress-ambiguous-set", + ), + ], + ) + def test_parse_command_line_flags( + self, + db: DatabaseTransactionFixture, + extra_args: list[str], + attribute: str, + expected: bool, + ): parsed = SuppressWorkForLibraryScript.parse_command_line( db.session, - ["--library", "lib1", "--identifier", "123", "--dry-run"], + ["--library", "lib1", "--identifier", "123", *extra_args], ) - assert parsed.dry_run is True + assert getattr(parsed, attribute) is expected def test_parse_command_line_file_and_identifier_mutually_exclusive( self, db: DatabaseTransactionFixture, capsys @@ -115,117 +191,73 @@ def test_load_identifier(self, db: DatabaseTransactionFixture): with pytest.raises(ValueError): script.load_identifier("test", "test") - def test_load_identifiers_from_file(self, db: DatabaseTransactionFixture, tmp_path): - csv_file = tmp_path / "ids.csv" - csv_file.write_text( - textwrap.dedent( + @pytest.mark.parametrize( + "csv_content,expected", + [ + pytest.param( """\ identifier,identifier_type 978-0-06-112008-4,ISBN 12345,Overdrive ID ,ISBN - """ - ) - ) - - script = SuppressWorkForLibraryScript(db.session) - pairs = script.load_identifiers_from_file(str(csv_file), "ISBN") - - assert pairs == [ - ("ISBN", "978-0-06-112008-4"), - ("Overdrive ID", "12345"), - ] - - def test_load_identifiers_from_file_no_type_column( - self, db: DatabaseTransactionFixture, tmp_path - ): - csv_file = tmp_path / "ids.csv" - csv_file.write_text( - textwrap.dedent( + """, + [("ISBN", "978-0-06-112008-4"), ("Overdrive ID", "12345")], + id="row-without-an-identifier-is-skipped", + ), + pytest.param( """\ identifier 978-0-06-112008-4 12345 - """ - ) - ) - - script = SuppressWorkForLibraryScript(db.session) - pairs = script.load_identifiers_from_file(str(csv_file), "ISBN") - - assert pairs == [ - ("ISBN", "978-0-06-112008-4"), - ("ISBN", "12345"), - ] - - def test_load_identifiers_from_file_omitted_type_value_falls_back_to_default( - self, db: DatabaseTransactionFixture, tmp_path - ): - """When identifier_type column exists but a row omits the value (e.g. '12345' - instead of '12345,'), DictReader sets it to None. We must not call .strip() - on None.""" - csv_file = tmp_path / "ids.csv" - csv_file.write_text( - textwrap.dedent( + """, + [("ISBN", "978-0-06-112008-4"), ("ISBN", "12345")], + id="no-type-column-falls-back-to-default", + ), + pytest.param( + # A row that omits the trailing comma entirely ('12345' rather + # than '12345,') leaves DictReader with None, not "", so the + # fallback must not call .strip() on it. """\ identifier,identifier_type 978-0-06-112008-4 12345,Overdrive ID - """ - ) - ) - - script = SuppressWorkForLibraryScript(db.session) - pairs = script.load_identifiers_from_file(str(csv_file), "ISBN") - - assert pairs == [ - ("ISBN", "978-0-06-112008-4"), - ("Overdrive ID", "12345"), - ] - - def test_load_identifiers_from_file_empty_type_falls_back_to_default( - self, db: DatabaseTransactionFixture, tmp_path - ): - csv_file = tmp_path / "ids.csv" - csv_file.write_text( - textwrap.dedent( + """, + [("ISBN", "978-0-06-112008-4"), ("Overdrive ID", "12345")], + id="omitted-type-value-falls-back-to-default", + ), + pytest.param( """\ identifier,identifier_type 978-0-06-112008-4, 12345,Overdrive ID - """ - ) - ) - - script = SuppressWorkForLibraryScript(db.session) - pairs = script.load_identifiers_from_file(str(csv_file), "ISBN") - - assert pairs == [ - ("ISBN", "978-0-06-112008-4"), - ("Overdrive ID", "12345"), - ] - - def test_load_identifiers_from_file_with_duplicates( - self, db: DatabaseTransactionFixture, tmp_path - ): - csv_file = tmp_path / "ids.csv" - csv_file.write_text( - textwrap.dedent( + """, + [("ISBN", "978-0-06-112008-4"), ("Overdrive ID", "12345")], + id="empty-type-value-falls-back-to-default", + ), + pytest.param( """\ identifier,identifier_type 978-0-06-112008-4,ISBN 978-0-06-112008-4,ISBN - """ - ) - ) + """, + [("ISBN", "978-0-06-112008-4"), ("ISBN", "978-0-06-112008-4")], + id="duplicates-are-preserved-for-do-run-to-dedupe", + ), + ], + ) + def test_load_identifiers_from_file( + self, + db: DatabaseTransactionFixture, + tmp_path, + csv_content: str, + expected: list[tuple[str, str]], + ): + csv_file = tmp_path / "ids.csv" + csv_file.write_text(textwrap.dedent(csv_content)) script = SuppressWorkForLibraryScript(db.session) - pairs = script.load_identifiers_from_file(str(csv_file), "ISBN") - assert pairs == [ - ("ISBN", "978-0-06-112008-4"), - ("ISBN", "978-0-06-112008-4"), - ] + assert script.load_identifiers_from_file(str(csv_file), "ISBN") == expected def test_do_run_deduplicates_and_warns( self, db: DatabaseTransactionFixture, tmp_path, caplog @@ -235,26 +267,17 @@ def test_do_run_deduplicates_and_warns( import logging test_library = db.library(short_name="test") - work1 = db.work(with_license_pool=True) - work2 = db.work(with_license_pool=True) - id1 = work1.presentation_edition.primary_identifier - id2 = work2.presentation_edition.primary_identifier - - csv_file = tmp_path / "ids.csv" - csv_file.write_text( - f"identifier,identifier_type\n" - f"{id1.identifier},{id1.type}\n" - f"{id1.identifier},{id1.type}\n" - f"{id2.identifier},{id2.type}\n" + works, csv_path = works_with_identifier_csv( + db, test_library, tmp_path, repeat_first=True ) caplog.set_level(logging.WARNING) script = SuppressWorkForLibraryScript(db.session) - script.do_run(["--library", test_library.short_name, "--file", str(csv_file)]) + script.do_run(["--library", test_library.short_name, "--file", csv_path]) assert "Removed 1 duplicate identifier(s) from input" in caplog.text - assert test_library in work1.suppressed_for - assert test_library in work2.suppressed_for + for work in works: + assert test_library in work.suppressed_for def test_load_identifiers_from_file_missing_identifier_column( self, db: DatabaseTransactionFixture, tmp_path @@ -266,204 +289,725 @@ def test_load_identifiers_from_file_missing_identifier_column( with pytest.raises(ValueError, match='must contain an "identifier" column'): script.load_identifiers_from_file(str(csv_file), "ISBN") - def test_do_run(self, db: DatabaseTransactionFixture, capsys): + @pytest.mark.parametrize( + "extra_args,expected_kwargs", + [ + pytest.param( + [], {"dry_run": False, "suppress_ambiguous": False}, id="no-flags" + ), + pytest.param( + ["--dry-run"], + {"dry_run": True, "suppress_ambiguous": False}, + id="dry-run", + ), + pytest.param( + ["--suppress-ambiguous"], + {"dry_run": False, "suppress_ambiguous": True}, + id="suppress-ambiguous", + ), + ], + ) + def test_do_run_passes_flags_to_suppress_work( + self, + db: DatabaseTransactionFixture, + capsys, + extra_args: list[str], + expected_kwargs: dict[str, bool], + ): test_library = db.library(short_name="test") test_identifier = db.identifier() script = SuppressWorkForLibraryScript(db.session) suppress_work_mock = create_autospec(script.suppress_work) - suppress_work_mock.return_value = SuppressResult.NEWLY_SUPPRESSED + suppress_work_mock.return_value = SuppressOutcome( + SuppressResult.NEWLY_SUPPRESSED, "Some Title" + ) script.suppress_work = suppress_work_mock - args = [ - "--library", - test_library.short_name, - "--identifier-type", - test_identifier.type, - "--identifier", - test_identifier.identifier, - ] - script.do_run(args) - suppress_work_mock.assert_called_once_with( - test_library, test_identifier, dry_run=False + script.do_run( + [ + "--library", + test_library.short_name, + "--identifier-type", + test_identifier.type, + "--identifier", + test_identifier.identifier, + *extra_args, + ] ) - def test_do_run_dry_run(self, db: DatabaseTransactionFixture, capsys): - test_library = db.library(short_name="test") - test_identifier = db.identifier() - - script = SuppressWorkForLibraryScript(db.session) - suppress_work_mock = create_autospec(script.suppress_work) - suppress_work_mock.return_value = SuppressResult.NEWLY_SUPPRESSED - script.suppress_work = suppress_work_mock - args = [ - "--library", - test_library.short_name, - "--identifier-type", - test_identifier.type, - "--identifier", - test_identifier.identifier, - "--dry-run", - ] - script.do_run(args) - suppress_work_mock.assert_called_once_with( - test_library, test_identifier, dry_run=True + test_library, test_identifier, **expected_kwargs ) def test_do_run_with_file(self, db: DatabaseTransactionFixture, tmp_path, capsys): test_library = db.library(short_name="test") - work1 = db.work(with_license_pool=True) - work2 = db.work(with_license_pool=True) - id1 = work1.presentation_edition.primary_identifier - id2 = work2.presentation_edition.primary_identifier - - csv_file = tmp_path / "ids.csv" - csv_file.write_text( - f"identifier,identifier_type\n" - f"{id1.identifier},{id1.type}\n" - f"{id2.identifier},{id2.type}\n" - ) + works, csv_path = works_with_identifier_csv(db, test_library, tmp_path) script = SuppressWorkForLibraryScript(db.session) - script.do_run( - [ - "--library", - test_library.short_name, - "--file", - str(csv_file), - ] - ) + script.do_run(["--library", test_library.short_name, "--file", csv_path]) - assert test_library in work1.suppressed_for - assert test_library in work2.suppressed_for + for work in works: + assert test_library in work.suppressed_for out = capsys.readouterr().out assert re.search(r"Newly suppressed:\s+2", out) assert re.search(r"Already suppressed:\s+0", out) assert re.search(r"Not found:\s+0", out) - def test_suppress_work(self, db: DatabaseTransactionFixture): + @pytest.mark.parametrize( + "already_suppressed,dry_run,expected_result,suppressed_after", + [ + pytest.param( + False, False, SuppressResult.NEWLY_SUPPRESSED, True, id="suppresses" + ), + pytest.param( + True, + False, + SuppressResult.ALREADY_SUPPRESSED, + True, + id="already-suppressed", + ), + pytest.param( + False, True, SuppressResult.NEWLY_SUPPRESSED, False, id="dry-run" + ), + pytest.param( + True, + True, + SuppressResult.ALREADY_SUPPRESSED, + True, + id="dry-run-already-suppressed", + ), + ], + ) + def test_suppress_work( + self, + db: DatabaseTransactionFixture, + already_suppressed: bool, + dry_run: bool, + expected_result: SuppressResult, + suppressed_after: bool, + ): test_library = db.library(short_name="test") - work = db.work(with_license_pool=True) - - assert work.suppressed_for == [] + collection = db.collection(library=test_library) + work = db.work(with_license_pool=True, collection=collection) + if already_suppressed: + work.suppressed_for.append(test_library) script = SuppressWorkForLibraryScript(db.session) result = script.suppress_work( - test_library, work.presentation_edition.primary_identifier + test_library, + work.presentation_edition.primary_identifier, + dry_run=dry_run, ) - assert result == SuppressResult.NEWLY_SUPPRESSED - assert work.suppressed_for == [test_library] + assert result.result == expected_result + assert result.description == f"{work.title} (work id: {work.id})" + assert work.suppressed_for == ([test_library] if suppressed_after else []) - def test_suppress_work_already_suppressed(self, db: DatabaseTransactionFixture): + def test_suppress_work_no_work_for_identifier(self, db: DatabaseTransactionFixture): + test_library = db.library(short_name="test") + db.collection(library=test_library) + identifier = db.identifier() + + script = SuppressWorkForLibraryScript(db.session) + result = script.suppress_work(test_library, identifier) + + assert result.result == SuppressResult.NOT_FOUND + assert result.description is None + + def test_suppress_work_library_with_no_collections( + self, db: DatabaseTransactionFixture + ): + """A library that carries no collections carries no works, so + there is nothing for it to suppress -- but the work does exist, + so this is reported as not-in-this-library rather than not-found.""" test_library = db.library(short_name="test") work = db.work(with_license_pool=True) - work.suppressed_for.append(test_library) script = SuppressWorkForLibraryScript(db.session) result = script.suppress_work( test_library, work.presentation_edition.primary_identifier ) - assert result == SuppressResult.ALREADY_SUPPRESSED - assert work.suppressed_for == [test_library] + assert result.result == SuppressResult.NOT_IN_LIBRARY + assert work.suppressed_for == [] - def test_suppress_work_no_work_for_identifier(self, db: DatabaseTransactionFixture): + def test_library_collection_ids_are_cached(self, db: DatabaseTransactionFixture): + """load_works consults this once per identifier, so a --file run + would otherwise repeat the same query for every row.""" test_library = db.library(short_name="test") - identifier = db.identifier() + collection = db.collection(library=test_library) + + script = SuppressWorkForLibraryScript(db.session) + first = script._library_collection_ids(test_library) + second = script._library_collection_ids(test_library) + + assert first == [collection.id] + assert first is second + + def test_suppress_work_pool_without_a_work(self, db: DatabaseTransactionFixture): + """A LicensePool can exist before its Work has been calculated + (work_id is nullable), so there may be nothing to suppress even + though the identifier is licensed by the library.""" + test_library = db.library(short_name="test") + collection = db.collection(library=test_library) + edition = db.edition() + db.licensepool(edition, collection=collection) + + script = SuppressWorkForLibraryScript(db.session) + result = script.suppress_work(test_library, edition.primary_identifier) + + assert result.result == SuppressResult.NOT_FOUND + + def test_suppress_work_pool_without_a_work_falls_through_to_equivalency( + self, db: DatabaseTransactionFixture + ): + """A pool whose Work hasn't been calculated can't be suppressed -- + there is no Work to attach the suppression to -- so it doesn't + count as an exact match and resolution falls through to + equivalency. Refusing instead would leave a title the library + demonstrably carries unsuppressed just because some other pool of + the same identifier was mid-import.""" + test_library = db.library(short_name="test") + collection = db.collection(library=test_library) + + # The identifier the librarian names is licensed here, but its + # pool has no Work yet. + workless_edition = db.edition() + db.licensepool(workless_edition, collection=collection) + identifier = workless_edition.primary_identifier + + # An equivalent identifier does have a Work in the same collection. + equivalent_work = db.work(with_license_pool=True, collection=collection) + source = DataSource.lookup(db.session, DataSource.OCLC) + identifier.equivalent_to( + source, equivalent_work.presentation_edition.primary_identifier, 1 + ) script = SuppressWorkForLibraryScript(db.session) result = script.suppress_work(test_library, identifier) - assert result == SuppressResult.NOT_FOUND + assert result.result == SuppressResult.NEWLY_SUPPRESSED + assert equivalent_work.suppressed_for == [test_library] - def test_suppress_work_dry_run(self, db: DatabaseTransactionFixture): + def test_suppress_work_not_in_library_distinguished_from_not_found( + self, db: DatabaseTransactionFixture + ): + """An identifier belonging to some other library's collection is a + different problem from an identifier that matches nothing at all, + so the two get distinct results instead of both reading NOT FOUND.""" test_library = db.library(short_name="test") - work = db.work(with_license_pool=True) + db.collection(library=test_library) + other_collection = db.collection() + work = db.work(with_license_pool=True, collection=other_collection) script = SuppressWorkForLibraryScript(db.session) result = script.suppress_work( - test_library, - work.presentation_edition.primary_identifier, - dry_run=True, + test_library, work.presentation_edition.primary_identifier ) - assert result == SuppressResult.NEWLY_SUPPRESSED + assert result.result == SuppressResult.NOT_IN_LIBRARY + # The work is named, so an operator can see what they don't carry. + assert result.description == f"{work.title} (work id: {work.id})" assert work.suppressed_for == [] - def test_suppress_work_dry_run_already_suppressed( + def test_suppress_work_resolves_via_equivalent_identifier( self, db: DatabaseTransactionFixture ): + """A librarian will typically have an ISBN in hand, but the + LicensePool is often keyed on a vendor identifier (e.g. an + Overdrive ID) with the ISBN linked only via identifier + equivalency. The script must resolve the work through that + equivalency instead of requiring the ISBN to be the + LicensePool's own identifier.""" test_library = db.library(short_name="test") - work = db.work(with_license_pool=True) - work.suppressed_for.append(test_library) + isbn, (work,) = isbn_equivalent_to_works(db, test_library, work_count=1) + + script = SuppressWorkForLibraryScript(db.session) + result = script.suppress_work(test_library, isbn) + + assert result.result == SuppressResult.NEWLY_SUPPRESSED + assert result.description == f"{work.title} (work id: {work.id})" + assert work.suppressed_for == [test_library] + + @pytest.mark.parametrize( + "strength,expected_result,expect_suppressed", + [ + pytest.param( + 1, SuppressResult.NEWLY_SUPPRESSED, True, id="full-confidence-resolves" + ), + pytest.param( + 0.85, SuppressResult.NOT_FOUND, False, id="below-threshold-ignored" + ), + ], + ) + def test_suppress_work_only_high_confidence_equivalencies_resolve( + self, + db: DatabaseTransactionFixture, + strength: float, + expected_result: SuppressResult, + expect_suppressed: bool, + ): + """Equivalency resolution uses `Work.from_identifiers`' strict + default policy (threshold 0.999), so only assertions a data source + is fully confident about can pull a work into a suppression. + + 0.85 isn't an arbitrary "low" number: it's the strength the + importer itself assigns when it links two identifiers purely + because their editions share a permanent work id + (`BibliographicData` in `data_layer/bibliographic.py`). Those + edges exist throughout production data, and a suppression must + not ride one into a work the librarian never named.""" + test_library = db.library(short_name="test") + collection = db.collection(library=test_library) + work = db.work(with_license_pool=True, collection=collection) + + isbn = db.identifier(identifier_type="ISBN") + source = DataSource.lookup(db.session, DataSource.OCLC) + isbn.equivalent_to( + source, work.presentation_edition.primary_identifier, strength + ) + + script = SuppressWorkForLibraryScript(db.session) + result = script.suppress_work(test_library, isbn) + + assert result.result == expected_result + assert work.suppressed_for == ([test_library] if expect_suppressed else []) + + def test_suppress_work_equivalent_identifier_only_affects_specified_library( + self, db: DatabaseTransactionFixture + ): + """Resolving the work via identifier equivalency must never + broaden *which libraries* get the work suppressed -- only the + library explicitly passed in should end up in + `work.suppressed_for`.""" + library_a = db.library(short_name="lib_a") + library_b = db.library(short_name="lib_b") + isbn, (work,) = isbn_equivalent_to_works(db, library_a, work_count=1) + + script = SuppressWorkForLibraryScript(db.session) + result = script.suppress_work(library_a, isbn) + + assert result.result == SuppressResult.NEWLY_SUPPRESSED + assert work.suppressed_for == [library_a] + assert library_b not in work.suppressed_for + + def test_suppress_work_ambiguous_equivalent_identifier( + self, db: DatabaseTransactionFixture + ): + """If an identifier resolves to more than one distinct Work via + equivalency, the script must not guess -- it should report + AMBIGUOUS and suppress nothing.""" + test_library = db.library(short_name="test") + isbn, (work1, work2) = isbn_equivalent_to_works(db, test_library) + + script = SuppressWorkForLibraryScript(db.session) + result = script.suppress_work(test_library, isbn) + + assert result.result == SuppressResult.AMBIGUOUS + # Candidates are ordered by work id, so the operator-facing output + # is stable between runs on the same data. + assert result.description == ( + f"{work1.title} (work id: {work1.id}); " + f"{work2.title} (work id: {work2.id})" + ) + assert work1.suppressed_for == [] + assert work2.suppressed_for == [] + + def test_suppress_work_not_ambiguous_when_only_one_candidate_is_licensed_to_library( + self, db: DatabaseTransactionFixture + ): + """Two different libraries each carrying "the same" title through + their own collection (e.g. library A via OverDrive, library B via + Bibliotheca) legitimately produces two distinct works reachable + from one shared ISBN. Suppressing for library A alone must resolve + cleanly to A's own work rather than reporting AMBIGUOUS over a + candidate that isn't even licensed to A.""" + library_a = db.library(short_name="lib_a") + library_b = db.library(short_name="lib_b") + collection_a = db.collection(library=library_a) + collection_b = db.collection(library=library_b) + work_a = db.work(with_license_pool=True, collection=collection_a) + work_b = db.work(with_license_pool=True, collection=collection_b) + id_a = work_a.presentation_edition.primary_identifier + id_b = work_b.presentation_edition.primary_identifier + + isbn = db.identifier(identifier_type="ISBN") + source = DataSource.lookup(db.session, DataSource.OCLC) + isbn.equivalent_to(source, id_a, 1) + isbn.equivalent_to(source, id_b, 1) + + script = SuppressWorkForLibraryScript(db.session) + result = script.suppress_work(library_a, isbn) + + assert result.result == SuppressResult.NEWLY_SUPPRESSED + assert result.description == f"{work_a.title} (work id: {work_a.id})" + assert work_a.suppressed_for == [library_a] + assert work_b.suppressed_for == [] + + def test_suppress_work_work_with_pools_in_several_collections( + self, db: DatabaseTransactionFixture + ): + """A Work can own pools in more than one collection (open-access + pools are merged across collections by permanent work id). The + library's own pool and the pool carrying the equivalent identifier + may therefore be different pools of the same Work, so the + collection scope has to be independent of the equivalency match + rather than riding on the same joined row.""" + test_library = db.library(short_name="test") + library_collection = db.collection(library=test_library) + other_collection = db.collection() + + work = db.work(with_license_pool=True, collection=library_collection) + + # A second pool of the same work, in a collection this library + # doesn't carry. The ISBN is equivalent to *this* pool's identifier. + other_edition = db.edition() + db.licensepool(other_edition, collection=other_collection, work=work) + + isbn = db.identifier(identifier_type="ISBN") + source = DataSource.lookup(db.session, DataSource.OCLC) + isbn.equivalent_to(source, other_edition.primary_identifier, 1) + + script = SuppressWorkForLibraryScript(db.session) + result = script.suppress_work(test_library, isbn) + + assert result.result == SuppressResult.NEWLY_SUPPRESSED + assert work.suppressed_for == [test_library] + + @pytest.mark.parametrize( + "suppress_ambiguous,expected_result,suppressed_after", + [ + pytest.param( + False, SuppressResult.AMBIGUOUS, False, id="refuses-without-flag" + ), + pytest.param( + True, SuppressResult.NEWLY_SUPPRESSED, True, id="covers-both-with-flag" + ), + ], + ) + def test_suppress_work_same_identifier_pointing_at_inconsistent_works( + self, + db: DatabaseTransactionFixture, + suppress_ambiguous: bool, + expected_result: SuppressResult, + suppressed_after: bool, + ): + """`LicensePool.calculate_work` forces every pool sharing an + identifier onto one Work, so two pools of one identifier pointing + at different Works is precisely the inconsistency it warns about + ("more than one Work between them") and repairs -- which is why + this state has to be built by hand here rather than through the + normal import path. + + The guard exists so that if the script does meet that state it + refuses, rather than suppressing one Work, reporting success, and + leaving the other circulating.""" + test_library = db.library(short_name="test") + edition = db.edition() + + # Two pools of one identifier, deliberately pointed at different + # works: the inconsistent state calculate_work would repair. + works = [] + for _ in range(2): + work = db.work(with_license_pool=False) + db.licensepool( + edition, collection=db.collection(library=test_library), work=work + ) + works.append(work) script = SuppressWorkForLibraryScript(db.session) result = script.suppress_work( test_library, - work.presentation_edition.primary_identifier, - dry_run=True, + edition.primary_identifier, + suppress_ambiguous=suppress_ambiguous, ) - assert result == SuppressResult.ALREADY_SUPPRESSED + assert result.result == expected_result + for work in works: + assert work.suppressed_for == ([test_library] if suppressed_after else []) + + def test_suppress_work_identifier_pool_outside_library_other_pool_inside( + self, db: DatabaseTransactionFixture + ): + """The identifier's own pool may sit outside the library while the + same Work has another pool inside it. Resolution has to find that + work through the equivalency query -- which reaches it because an + identifier is always a member of its own equivalent set (the + fn_recursive_equivalents base case seeds the CTE with it).""" + test_library = db.library(short_name="test") + library_collection = db.collection(library=test_library) + other_collection = db.collection() + + # The work's pool for `identifier` is in a collection the library + # doesn't carry... + work = db.work(with_license_pool=True, collection=other_collection) + identifier = work.presentation_edition.primary_identifier + + # ...but the same work has another pool in a collection it does. + inside_edition = db.edition() + db.licensepool(inside_edition, collection=library_collection, work=work) - def test_print_results_normal(self, db: DatabaseTransactionFixture, capsys): - test_library = db.library(short_name="mylib", name="My Library") script = SuppressWorkForLibraryScript(db.session) - results = { - ("ISBN", "111"): SuppressResult.NEWLY_SUPPRESSED, - ("ISBN", "222"): SuppressResult.ALREADY_SUPPRESSED, - ("ISBN", "333"): SuppressResult.NOT_FOUND, - } - started_at = datetime(2026, 2, 26, 12, 0, 0, tzinfo=timezone.utc) - script._print_results( - results, - dry_run=False, - library=test_library, - started_at=started_at, - duration_seconds=1.23, + result = script.suppress_work(test_library, identifier) + + assert result.result == SuppressResult.NEWLY_SUPPRESSED + assert result.description == f"{work.title} (work id: {work.id})" + assert work.suppressed_for == [test_library] + + def test_suppress_work_direct_match_outside_library_falls_through( + self, db: DatabaseTransactionFixture + ): + """An ISBN can be some *other* library's collection's own pool + identifier (ISBN-keyed ODL/OPDS collections are the common case). + That direct match isn't this library's work, so it must not be + suppressed on this library's behalf -- and the equivalency + fallback should still find the work this library does carry.""" + test_library = db.library(short_name="test") + library_collection = db.collection(library=test_library) + other_collection = db.collection() + + work = db.work(with_license_pool=True, collection=library_collection) + + # An ISBN that is another collection's own pool identifier. + isbn_edition = db.edition(identifier_type="ISBN") + other_work = db.work(with_license_pool=True, collection=other_collection) + db.licensepool(isbn_edition, collection=other_collection, work=other_work) + isbn = isbn_edition.primary_identifier + + source = DataSource.lookup(db.session, DataSource.OCLC) + isbn.equivalent_to(source, work.presentation_edition.primary_identifier, 1) + + script = SuppressWorkForLibraryScript(db.session) + result = script.suppress_work(test_library, isbn) + + assert result.result == SuppressResult.NEWLY_SUPPRESSED + assert result.description == f"{work.title} (work id: {work.id})" + assert work.suppressed_for == [test_library] + assert other_work.suppressed_for == [] + + def test_suppress_work_suppress_ambiguous_suppresses_all_candidates( + self, db: DatabaseTransactionFixture + ): + """With --suppress-ambiguous, an identifier resolving to multiple + distinct works (e.g. the same ISBN licensed through more than one + collection) should suppress the work for the library in every + candidate, rather than refusing.""" + test_library = db.library(short_name="test") + isbn, (work1, work2) = isbn_equivalent_to_works(db, test_library) + + script = SuppressWorkForLibraryScript(db.session) + result = script.suppress_work(test_library, isbn, suppress_ambiguous=True) + + assert result.result == SuppressResult.NEWLY_SUPPRESSED + assert result.description == ( + f"{work1.title} (work id: {work1.id}); " + f"{work2.title} (work id: {work2.id})" ) + assert work1.suppressed_for == [test_library] + assert work2.suppressed_for == [test_library] - out = capsys.readouterr().out - assert "Suppression Results Summary" in out - assert "My Library (mylib)" in out - assert "2026-02-26 12:00:00 UTC" in out - assert "1.23s" in out - assert re.search(r"Newly suppressed:\s+1", out) - assert re.search(r"Already suppressed:\s+1", out) - assert re.search(r"Not found:\s+1", out) - assert "[SUPPRESSED] ISBN/111" in out - assert "[ALREADY SUPPRESSED] ISBN/222" in out - assert "[NOT FOUND] ISBN/333" in out - assert "[DRY RUN]" not in out + @pytest.mark.parametrize( + "suppress_ambiguous", + [pytest.param(False, id="without-flag"), pytest.param(True, id="with-flag")], + ) + def test_suppress_work_all_candidates_already_suppressed( + self, db: DatabaseTransactionFixture, suppress_ambiguous: bool + ): + """Re-running against a fully-covered set of candidates (e.g. + after an earlier --suppress-ambiguous run) must be idempotent: it + should report ALREADY_SUPPRESSED rather than AMBIGUOUS whether or + not the flag is passed, since there's nothing left to decide or + change -- and no duplicate rows from a redundant append.""" + test_library = db.library(short_name="test") + isbn, works = isbn_equivalent_to_works(db, test_library) + for work in works: + work.suppressed_for.append(test_library) + + script = SuppressWorkForLibraryScript(db.session) + result = script.suppress_work( + test_library, isbn, suppress_ambiguous=suppress_ambiguous + ) + + assert result.result == SuppressResult.ALREADY_SUPPRESSED + for work in works: + assert work.suppressed_for == [test_library] + + def test_suppress_work_suppress_ambiguous_partial_already_suppressed( + self, db: DatabaseTransactionFixture + ): + """If some but not all candidates are already suppressed, the + overall result is NEWLY_SUPPRESSED since the run had an effect, + every candidate ends up suppressed, and the reported title + describes only the candidate that actually changed.""" + test_library = db.library(short_name="test") + isbn, (work1, work2) = isbn_equivalent_to_works(db, test_library) + work1.suppressed_for.append(test_library) + + script = SuppressWorkForLibraryScript(db.session) + result = script.suppress_work(test_library, isbn, suppress_ambiguous=True) + + assert result.result == SuppressResult.NEWLY_SUPPRESSED + assert work1.suppressed_for == [test_library] + assert work2.suppressed_for == [test_library] + # Only the newly-changed work2 is described -- work1 was already + # suppressed before this run, so it isn't reported as an action + # that just happened. + assert result.description == f"{work2.title} (work id: {work2.id})" + + def test_suppress_work_suppress_ambiguous_dry_run_does_not_mutate( + self, db: DatabaseTransactionFixture + ): + test_library = db.library(short_name="test") + isbn, works = isbn_equivalent_to_works(db, test_library) + + script = SuppressWorkForLibraryScript(db.session) + result = script.suppress_work( + test_library, isbn, dry_run=True, suppress_ambiguous=True + ) + + assert result.result == SuppressResult.NEWLY_SUPPRESSED + for work in works: + assert work.suppressed_for == [] + + def test_suppress_work_prefers_direct_match_over_ambiguous_equivalency( + self, db: DatabaseTransactionFixture + ): + """An identifier that owns a LicensePool directly is an exact + match and must win even if it's also equivalent (via messy + metadata) to a different, unrelated Work -- equivalency should + only be consulted as a fallback when there's no direct match, + never used to second-guess one.""" + test_library = db.library(short_name="test") + collection = db.collection(library=test_library) + work = db.work(with_license_pool=True, collection=collection) + identifier = work.presentation_edition.primary_identifier + + other_work = db.work(with_license_pool=True, collection=collection) + other_identifier = other_work.presentation_edition.primary_identifier + + source = DataSource.lookup(db.session, DataSource.OCLC) + identifier.equivalent_to(source, other_identifier, 1) + + script = SuppressWorkForLibraryScript(db.session) + result = script.suppress_work(test_library, identifier) + + assert result.result == SuppressResult.NEWLY_SUPPRESSED + assert result.description == f"{work.title} (work id: {work.id})" + assert work.suppressed_for == [test_library] + assert other_work.suppressed_for == [] - def test_print_results_dry_run(self, db: DatabaseTransactionFixture, capsys): + @pytest.mark.parametrize( + "dry_run,results,expected,absent", + [ + pytest.param( + False, + { + ("ISBN", "111"): SuppressOutcome( + SuppressResult.NEWLY_SUPPRESSED, "Book One" + ), + ("ISBN", "222"): SuppressOutcome( + SuppressResult.ALREADY_SUPPRESSED, "Book Two" + ), + ("ISBN", "333"): SuppressOutcome(SuppressResult.NOT_FOUND), + }, + [ + "Suppression Results Summary", + "My Library (mylib)", + "2026-02-26 12:00:00 UTC", + "1.23s", + "Newly suppressed: 1", + "Already suppressed: 1", + "Not found: 1", + "[SUPPRESSED] ISBN/111 -- Book One", + "[ALREADY SUPPRESSED] ISBN/222 -- Book Two", + "[NOT FOUND] ISBN/333", + ], + ["[DRY RUN]"], + id="normal", + ), + pytest.param( + True, + { + ("ISBN", "111"): SuppressOutcome( + SuppressResult.NEWLY_SUPPRESSED, "Book One" + ), + ("ISBN", "222"): SuppressOutcome(SuppressResult.NOT_FOUND), + }, + [ + "[DRY RUN] Suppression Results Summary", + "Would suppress: 1", + "Not found: 1", + "[WOULD SUPPRESS] ISBN/111 -- Book One", + "[NOT FOUND] ISBN/222", + ], + ["[SUPPRESSED]"], + id="dry-run", + ), + pytest.param( + False, + { + ("ISBN", "111"): SuppressOutcome( + SuppressResult.AMBIGUOUS, "Book One; Book Two" + ), + }, + [ + "Ambiguous: 1", + "[AMBIGUOUS] ISBN/111 -- Book One; Book Two", + ], + [], + id="ambiguous", + ), + pytest.param( + False, + { + ("ISBN", "111"): SuppressOutcome( + SuppressResult.NOT_IN_LIBRARY, "Book One (work id: 1)" + ), + ("ISBN", "222"): SuppressOutcome(SuppressResult.NOT_FOUND), + }, + # The two misses are counted separately, so an operator can + # tell a title they don't carry from an identifier that + # matches nothing. + [ + "Not in this library: 1", + "Not found: 1", + "[NOT IN THIS LIBRARY] ISBN/111 -- Book One (work id: 1)", + "[NOT FOUND] ISBN/222", + ], + [], + id="not-in-library", + ), + ], + ) + def test_print_results( + self, + db: DatabaseTransactionFixture, + capsys, + dry_run: bool, + results: dict[tuple[str, str], SuppressOutcome], + expected: list[str], + absent: list[str], + ): test_library = db.library(short_name="mylib", name="My Library") script = SuppressWorkForLibraryScript(db.session) - results = { - ("ISBN", "111"): SuppressResult.NEWLY_SUPPRESSED, - ("ISBN", "222"): SuppressResult.NOT_FOUND, - } - started_at = datetime(2026, 2, 26, 9, 30, 0, tzinfo=timezone.utc) + script._print_results( results, - dry_run=True, + dry_run=dry_run, library=test_library, - started_at=started_at, - duration_seconds=0.05, + started_at=datetime(2026, 2, 26, 12, 0, 0, tzinfo=timezone.utc), + duration_seconds=1.23, ) - out = capsys.readouterr().out - assert "[DRY RUN] Suppression Results Summary" in out - assert "My Library (mylib)" in out - assert "2026-02-26 09:30:00 UTC" in out - assert "0.05s" in out - assert re.search(r"Would suppress:\s+1", out) - assert re.search(r"Not found:\s+1", out) - assert "[WOULD SUPPRESS] ISBN/111" in out - assert "[NOT FOUND] ISBN/222" in out + # Summary rows are column-padded, so compare against a + # whitespace-collapsed copy to keep the expectations readable. + out = re.sub(r"\s+", " ", capsys.readouterr().out) + for fragment in expected: + assert fragment in out + for fragment in absent: + assert fragment not in out def test_do_run_not_found_identifier(self, db: DatabaseTransactionFixture, capsys): test_library = db.library(short_name="test") @@ -489,43 +1033,21 @@ def test_do_run_commits_once_for_all_suppressions( self, db: DatabaseTransactionFixture, tmp_path, capsys ): test_library = db.library(short_name="test") - work1 = db.work(with_license_pool=True) - work2 = db.work(with_license_pool=True) - id1 = work1.presentation_edition.primary_identifier - id2 = work2.presentation_edition.primary_identifier - - csv_file = tmp_path / "ids.csv" - csv_file.write_text( - f"identifier,identifier_type\n" - f"{id1.identifier},{id1.type}\n" - f"{id2.identifier},{id2.type}\n" - ) + works, csv_path = works_with_identifier_csv(db, test_library, tmp_path) script = SuppressWorkForLibraryScript(db.session) with patch.object(db.session, "commit", wraps=db.session.commit) as mock_commit: - script.do_run( - ["--library", test_library.short_name, "--file", str(csv_file)] - ) + script.do_run(["--library", test_library.short_name, "--file", csv_path]) mock_commit.assert_called_once() - assert test_library in work1.suppressed_for - assert test_library in work2.suppressed_for + for work in works: + assert test_library in work.suppressed_for def test_do_run_rolls_back_all_on_commit_failure( self, db: DatabaseTransactionFixture, tmp_path ): test_library = db.library(short_name="test") - work1 = db.work(with_license_pool=True) - work2 = db.work(with_license_pool=True) - id1 = work1.presentation_edition.primary_identifier - id2 = work2.presentation_edition.primary_identifier - - csv_file = tmp_path / "ids.csv" - csv_file.write_text( - f"identifier,identifier_type\n" - f"{id1.identifier},{id1.type}\n" - f"{id2.identifier},{id2.type}\n" - ) + _, csv_path = works_with_identifier_csv(db, test_library, tmp_path) script = SuppressWorkForLibraryScript(db.session) with ( @@ -534,7 +1056,7 @@ def test_do_run_rolls_back_all_on_commit_failure( ): with pytest.raises(Exception, match="DB error"): script.do_run( - ["--library", test_library.short_name, "--file", str(csv_file)] + ["--library", test_library.short_name, "--file", csv_path] ) mock_rollback.assert_called_once() @@ -593,12 +1115,13 @@ def test_load_identifiers_from_file_not_found(self, db: DatabaseTransactionFixtu def test_suppress_work_does_not_commit(self, db: DatabaseTransactionFixture): test_library = db.library(short_name="test") - work = db.work(with_license_pool=True) + collection = db.collection(library=test_library) + work = db.work(with_license_pool=True, collection=collection) script = SuppressWorkForLibraryScript(db.session) with patch.object(db.session, "commit") as mock_commit: result = script.suppress_work( test_library, work.presentation_edition.primary_identifier ) - assert result == SuppressResult.NEWLY_SUPPRESSED + assert result.result == SuppressResult.NEWLY_SUPPRESSED mock_commit.assert_not_called()