diff --git a/src/palace/manager/feed/annotator/circulation.py b/src/palace/manager/feed/annotator/circulation.py index 52499c4d97..0667c24af9 100644 --- a/src/palace/manager/feed/annotator/circulation.py +++ b/src/palace/manager/feed/annotator/circulation.py @@ -38,7 +38,9 @@ FeedData, IndirectAcquisition, Link, + LinkContentType, LinkKwargs, + LinkType, PatronData, WorkEntry, ) @@ -186,15 +188,15 @@ def license_tags( ) @classmethod - def format_types(cls, delivery_mechanism: DeliveryMechanism) -> list[str]: + def format_types(cls, delivery_mechanism: DeliveryMechanism) -> list[LinkType]: """Generate a set of types suitable for passing into acquisition_link(). """ - types = [] + types: list[LinkType] = [] # If this is a streaming book, you have to get an OPDS entry, then # get a direct link to the streaming reader from that. if delivery_mechanism.is_streaming: - types.append(OPDSFeed.ENTRY_TYPE) + types.append(LinkContentType.OPDS_ENTRY) # If this is a DRM-encrypted book, you have to get through the DRM # to get the goodies inside. @@ -624,7 +626,7 @@ def acquisition_link( cls, rel: str, href: str, - types: list[str] | None, + types: list[LinkType] | None, active_loan: Loan | None = None, templated: bool = False, ) -> Acquisition: @@ -649,7 +651,7 @@ def acquisition_link( @classmethod def indirect_acquisition( - cls, indirect_types: list[str] + cls, indirect_types: list[LinkType] ) -> IndirectAcquisition | None: top_level_parent: IndirectAcquisition | None = None parent: IndirectAcquisition | None = None @@ -737,8 +739,7 @@ def is_novelist_configured(self) -> bool: def top_level_title(self) -> str: return self._top_level_title - def permalink_for(self, identifier: Identifier) -> tuple[str, str]: - # TODO: Do not force OPDS types + def permalink_for(self, identifier: Identifier) -> tuple[str, LinkType]: url = self.url_for( "permalink", identifier_type=identifier.type, @@ -746,7 +747,7 @@ def permalink_for(self, identifier: Identifier) -> tuple[str, str]: library_short_name=self.library.short_name, _external=True, ) - return url, OPDSFeed.ENTRY_TYPE + return url, LinkContentType.OPDS_ENTRY def groups_url( self, lane: WorkList | None, facets: FacetsWithEntryPoint | None = None @@ -872,7 +873,6 @@ def annotate_work_entry( identifier = entry.identifier permalink_uri, permalink_type = self.permalink_for(identifier) - # TODO: Do not force OPDS types if permalink_uri: entry.computed.other_links.append( Link(href=permalink_uri, rel="alternate", type=permalink_type) @@ -895,7 +895,7 @@ def annotate_work_entry( entry.computed.other_links.append( Link( rel="recommendations", - type=OPDSFeed.ACQUISITION_FEED_TYPE, + type=LinkContentType.OPDS_FEED, title="Recommended Works", href=self.url_for( "recommendations", @@ -912,7 +912,7 @@ def annotate_work_entry( entry.computed.other_links.append( Link( rel="related", - type=OPDSFeed.ACQUISITION_FEED_TYPE, + type=LinkContentType.OPDS_FEED, title="Recommended Works", href=self.url_for( "related_books", @@ -1022,7 +1022,7 @@ def add_author_links(self, entry: WorkEntry) -> None: author_entry.link = Link( rel="contributor", - type=OPDSFeed.ACQUISITION_FEED_TYPE, + type=LinkContentType.OPDS_FEED, title=name, href=self.url_for( "contributor", @@ -1065,7 +1065,7 @@ def add_series_link(self, entry: WorkEntry) -> None: ) series_entry.link = Link( rel="series", - type=OPDSFeed.ACQUISITION_FEED_TYPE, + type=LinkContentType.OPDS_FEED, title=series_name, href=href, ) @@ -1118,7 +1118,7 @@ def annotate_feed(self, feed: FeedData) -> None: _external=True, ), rel="http://opds-spec.org/shelf", - type=OPDSFeed.ACQUISITION_FEED_TYPE, + type=LinkContentType.OPDS_FEED, ) feed.add_link( @@ -1151,7 +1151,7 @@ def annotate_feed(self, feed: FeedData) -> None: feed.add_link( href=crawlable_url, rel="http://opds-spec.org/crawlable", - type=OPDSFeed.ACQUISITION_FEED_TYPE, + type=LinkContentType.OPDS_FEED, ) self.add_configuration_links(feed) @@ -1358,7 +1358,7 @@ def borrow_link( borrow_link = Acquisition( rel=rel, href=borrow_url, - type=OPDSFeed.ENTRY_TYPE, + type=LinkContentType.OPDS_ENTRY, is_hold=True if active_hold else False, ) @@ -1421,7 +1421,9 @@ def fulfill_link( _external=True, ) - if template_vars := self.FULFILL_LINK_TEMPLATED_TYPES.get(format_types[0]): + first_format_type = format_types[0] + template_key = first_format_type if isinstance(first_format_type, str) else None + if template_vars := self.FULFILL_LINK_TEMPLATED_TYPES.get(template_key): fulfill_url = fulfill_url + "{?" + ",".join(template_vars) + "}" templated = True else: diff --git a/src/palace/manager/feed/annotator/loan_and_hold.py b/src/palace/manager/feed/annotator/loan_and_hold.py index fdb95233e4..0460b83fd2 100644 --- a/src/palace/manager/feed/annotator/loan_and_hold.py +++ b/src/palace/manager/feed/annotator/loan_and_hold.py @@ -57,7 +57,7 @@ def user_profile_management_protocol_link(self) -> Link: for the current patron. """ return Link( - rel="http://librarysimplified.org/terms/rel/user-profile", + rel="profile", href=self.url_for( "patron_profile", library_short_name=self.library.short_name, diff --git a/src/palace/manager/feed/serializer/opds.py b/src/palace/manager/feed/serializer/opds.py index eefb8bda70..56200360d9 100644 --- a/src/palace/manager/feed/serializer/opds.py +++ b/src/palace/manager/feed/serializer/opds.py @@ -5,8 +5,10 @@ from functools import partial from typing import Any, cast +from frozendict import frozendict from lxml import etree +from palace.manager.core.user_profile import ProfileController from palace.manager.feed.facets.constants import FacetConstants from palace.manager.feed.serializer.base import SerializerInterface from palace.manager.feed.types import ( @@ -18,6 +20,8 @@ FeedMetadata, IndirectAcquisition, Link, + LinkContentType, + LinkType, PatronData, Rating, Series, @@ -73,9 +77,36 @@ def is_sort_facet(link: Link) -> bool: class BaseOPDS1Serializer(SerializerInterface[etree._Element], OPDSFeed, abc.ABC): + _CONTENT_TYPE_MAP: frozendict[LinkContentType, str] = frozendict( + { + LinkContentType.OPDS_FEED: OPDSFeed.ACQUISITION_FEED_TYPE, + LinkContentType.OPDS_ENTRY: OPDSFeed.ENTRY_TYPE, + } + ) + + # OPDS1 uses Palace-specific relation URIs for some standard rels. + # OPDS2 uses standard IANA rels directly and needs no mapping. + _REL_MAP: frozendict[str, str] = frozendict( + { + "profile": ProfileController.LINK_RELATION, + } + ) + def __init__(self) -> None: pass + def _resolve_type(self, type_value: LinkType | None) -> str | None: + """Map semantic LinkContentType values to OPDS1-specific types.""" + if isinstance(type_value, LinkContentType): + return self._CONTENT_TYPE_MAP[type_value] + return type_value + + def _resolve_rel(self, rel_value: str | None) -> str | None: + """Map standard rels to OPDS1/Palace-specific rels.""" + if rel_value is not None and rel_value in self._REL_MAP: + return self._REL_MAP[rel_value] + return rel_value + def _tag( self, tag_name: str, *args: Any, mapping: dict[str, str] | None = None ) -> etree._Element: @@ -325,13 +356,15 @@ def _serialize_author_tag(self, tag: str, author: Author) -> etree._Element: return entry def _serialize_link(self, link: Link) -> etree._Element: + resolved_type = self._resolve_type(link.type) + resolved_rel = self._resolve_rel(link.rel) attrs: dict[str, str] = {} if link.href is not None: attrs["href"] = link.href - if link.rel is not None: - attrs["rel"] = link.rel - if link.type is not None: - attrs["type"] = link.type + if resolved_rel is not None: + attrs["rel"] = resolved_rel + if resolved_type is not None: + attrs["type"] = resolved_type if link.title is not None: attrs["title"] = link.title if link.role is not None: @@ -408,13 +441,23 @@ def _serialize_rating(self, rating: Rating) -> etree._Element: return entry def _serialize_acquisition_link(self, link: Acquisition) -> etree._Element: + resolved_type = self._resolve_type(link.type) + resolved_rel = self._resolve_rel(link.rel) + + attrs: dict[str, str] = {"href": link.href} + if resolved_rel is not None: + attrs["rel"] = resolved_rel + if resolved_type is not None: + attrs["type"] = resolved_type link_func = OPDSFeed.tlink if link.templated else OPDSFeed.link - element = link_func(**link.link_attribs()) + element = link_func(**attrs) def _indirect(item: IndirectAcquisition) -> etree._Element: tag = self._tag("indirectAcquisition") - tag.set("type", item.type) + resolved_indirect_type = self._resolve_type(item.type) + if resolved_indirect_type is not None: + tag.set("type", resolved_indirect_type) for child in item.children: tag.append(_indirect(child)) return tag diff --git a/src/palace/manager/feed/serializer/opds2.py b/src/palace/manager/feed/serializer/opds2.py index a581514844..3d1c999422 100644 --- a/src/palace/manager/feed/serializer/opds2.py +++ b/src/palace/manager/feed/serializer/opds2.py @@ -3,6 +3,7 @@ from collections.abc import Iterable from typing import Any +from frozendict import frozendict from pydantic import ValidationError from palace.manager.feed.serializer.base import SerializerInterface @@ -14,6 +15,8 @@ FeedData, IndirectAcquisition, Link, + LinkContentType, + LinkType, WorkEntryData, ) from palace.manager.opds import opds2, rwpm, schema_org @@ -47,6 +50,19 @@ class OPDS2Serializer(SerializerInterface[dict[str, Any]], LoggerMixin): + _CONTENT_TYPE_MAP: frozendict[LinkContentType, str] = frozendict( + { + LinkContentType.OPDS_FEED: opds2.Feed.content_type(), + LinkContentType.OPDS_ENTRY: opds2.BasePublication.content_type(), + } + ) + + def _resolve_type(self, type_value: LinkType | None) -> str | None: + """Map semantic LinkContentType values to OPDS2-specific types.""" + if isinstance(type_value, LinkContentType): + return self._CONTENT_TYPE_MAP[type_value] + return type_value + def serialize_feed( self, feed: FeedData, precomposed_entries: list[Any] | None = None ) -> str: @@ -201,14 +217,15 @@ def _serialize_publication_links( if link.rel is None: self.log.warning(f"Skipping OPDS2 link without rel: {link.href}") continue - if link.type is None: + resolved_type = self._resolve_type(link.type) + if resolved_type is None: self.log.error(f"Skipping OPDS2 link without type: {link.href}") continue links.append( self._strict_link( href=link.href, rel=link.rel, - type=link.type, + type=resolved_type, title=link.title, properties=self._link_properties(), ) @@ -224,7 +241,7 @@ def _serialize_link(self, link: Link) -> opds2.Link: return opds2.Link( href=link.href, rel=link.rel, - type=link.type, + type=self._resolve_type(link.type), title=link.title, ) @@ -290,7 +307,8 @@ def _serialize_acquisition_properties( def _serialize_indirect_acquisition( self, indirect: IndirectAcquisition ) -> opds2.AcquisitionObject | None: - if indirect.type is None: + indirect_type = self._resolve_type(indirect.type) + if indirect_type is None: self.log.error(f"Skipping indirect acquisition without type") return None children = [ @@ -299,7 +317,7 @@ def _serialize_indirect_acquisition( if (acq := self._serialize_indirect_acquisition(child)) is not None ] return opds2.AcquisitionObject( - type=indirect.type, + type=indirect_type, child=children, ) @@ -309,7 +327,7 @@ def _serialize_contributor(self, author: Author) -> rwpm.Contributor: rwpm.Link( href=link.href, rel=link.rel, - type=link.type, + type=self._resolve_type(link.type), ) ] if (link := author.link) and link.href @@ -342,10 +360,11 @@ def _serialize_feed_link(self, link: Link) -> opds2.StrictLink | None: if link.rel is None: self.log.warning(f"Skipping OPDS2 feed link without rel: {link.href}") return None + resolved_type = self._resolve_type(link.type) return self._strict_link( href=link.href, rel=link.rel, - type=link.type or self.content_type(), + type=resolved_type or self.content_type(), title=link.title, properties=self._link_properties(), ) @@ -372,7 +391,7 @@ def _serialize_facet_links(self, feed: FeedData) -> list[opds2.Facet]: href=link.href, title=title, rel=rel, - type=link.type, + type=self._resolve_type(link.type), properties=props, ) ) @@ -400,7 +419,7 @@ def _serialize_sort_link(self, link: Link) -> opds2.StrictLink: return self._strict_link( href=link.href, rel=PALACE_REL_SORT, - type=link.type or self.content_type(), + type=self._resolve_type(link.type) or self.content_type(), title=link.title, properties=self._link_properties( palace_active_sort=link.active_facet or None, @@ -420,7 +439,7 @@ def _serialize_navigation(self, feed: FeedData) -> list[opds2.TitleLink]: href=link.href, title=title, rel=link.rel, - type=link.type, + type=self._resolve_type(link.type), properties=self._link_properties(), ) ) @@ -428,10 +447,11 @@ def _serialize_navigation(self, feed: FeedData) -> list[opds2.TitleLink]: def _acquisition_link_type(self, link: Acquisition) -> str | None: if link.type: - return link.type + return self._resolve_type(link.type) for indirect in link.indirect_acquisitions: - if indirect.type: - return indirect.type + indirect_type = self._resolve_type(indirect.type) + if indirect_type: + return indirect_type self.log.error(f"Skipping acquisition link without type: {link.href}") return None diff --git a/src/palace/manager/feed/types.py b/src/palace/manager/feed/types.py index 8375a62fc6..a7aae37edc 100644 --- a/src/palace/manager/feed/types.py +++ b/src/palace/manager/feed/types.py @@ -4,7 +4,7 @@ from dataclasses import dataclass, field from datetime import date, datetime -from enum import StrEnum +from enum import Enum, StrEnum, auto from typing import Literal, NotRequired, TypedDict, Unpack from palace.manager.sqlalchemy.model.edition import Edition @@ -13,19 +13,27 @@ from palace.manager.sqlalchemy.model.work import Work -class LinkAttributes(TypedDict): - """Typed mapping for attributes used in OPDS1 link serialization.""" +class LinkContentType(Enum): + """Semantic content types for links that reference OPDS feeds or entries. - href: str - rel: NotRequired[str] - type: NotRequired[str] + Each serializer maps these to its format-specific content type. + Links with concrete types (text/html, application/json, etc.) bypass this. + """ + + OPDS_FEED = auto() + OPDS_ENTRY = auto() + + +#: Union type for link type fields that accept either concrete MIME types or +#: semantic :class:`LinkContentType` values resolved at serialization time. +LinkType = str | LinkContentType class LinkKwargs(TypedDict): """Typed keyword arguments accepted by FeedData.add_link.""" rel: NotRequired[str] - type: NotRequired[str] + type: NotRequired[LinkType] title: NotRequired[str] role: NotRequired[str] facet_group: NotRequired[str] @@ -49,7 +57,7 @@ class Link: href: str rel: str | None = None - type: str | None = None + type: LinkType | None = None # Additional types role: str | None = None @@ -62,15 +70,6 @@ class Link: default_facet: bool = False active_sort: bool = False - def link_attribs(self) -> LinkAttributes: - """Return the basic link attributes required for OPDS1.""" - attrs: LinkAttributes = {"href": self.href} - if self.rel is not None: - attrs["rel"] = self.rel - if self.type is not None: - attrs["type"] = self.type - return attrs - @dataclass(slots=True) class Category: @@ -127,7 +126,7 @@ class DRMLicensor: class IndirectAcquisition: """Tree structure for indirect acquisitions in OPDS1.""" - type: str | None = None + type: LinkType | None = None children: list[IndirectAcquisition] = field(default_factory=list) diff --git a/tests/manager/feed/test_library_annotator.py b/tests/manager/feed/test_library_annotator.py index 2ac04403dd..c0831b6188 100644 --- a/tests/manager/feed/test_library_annotator.py +++ b/tests/manager/feed/test_library_annotator.py @@ -31,7 +31,7 @@ from palace.manager.feed.facets.base import FacetsWithEntryPoint from palace.manager.feed.facets.feed import Facets from palace.manager.feed.opds import UnfulfillableWork -from palace.manager.feed.types import FeedData, WorkEntry +from palace.manager.feed.types import FeedData, LinkContentType, WorkEntry from palace.manager.feed.util import strftime from palace.manager.feed.worklist.contributor import ContributorLane from palace.manager.integration.goals import Goals @@ -525,7 +525,7 @@ def test_alternate_link_is_permalink( pool.identifier ) assert alternate == permalink - assert OPDSFeed.ENTRY_TYPE == type + assert LinkContentType.OPDS_ENTRY == type assert permalink_type == type # Make sure we are using the 'permalink' controller -- we were using @@ -825,7 +825,7 @@ def test_work_entry_includes_contributor_links( expected_rel_and_partial = dict(contributor="/contributor") self.assert_link_on_entry( entry.computed.authors, - link_type=OPDSFeed.ACQUISITION_FEED_TYPE, + link_type=LinkContentType.OPDS_FEED, partials_by_rel=expected_rel_and_partial, ) @@ -841,7 +841,7 @@ def test_work_entry_includes_contributor_links( assert 2 == len(contributor_links) contributor_links.sort(key=lambda l: l.href) for l in contributor_links: - assert l.type == OPDSFeed.ACQUISITION_FEED_TYPE + assert l.type == LinkContentType.OPDS_FEED assert "/contributor" in l.href assert contributor1.sort_name in contributor_links[0].href assert "Oprah" in contributor_links[1].href @@ -868,7 +868,7 @@ def test_work_entry_includes_series_link( expected_rel_and_partial = dict(series="/series") self.assert_link_on_entry( entry.computed.series, - link_type=OPDSFeed.ACQUISITION_FEED_TYPE, + link_type=LinkContentType.OPDS_FEED, partials_by_rel=expected_rel_and_partial, ) @@ -903,7 +903,7 @@ def test_work_entry_includes_recommendations_link( expected_rel_and_partial = dict(recommendations="/recommendations") self.assert_link_on_entry( entry, - link_type=OPDSFeed.ACQUISITION_FEED_TYPE, + link_type=LinkContentType.OPDS_FEED, partials_by_rel=expected_rel_and_partial, ) diff --git a/tests/manager/feed/test_opds2_serializer.py b/tests/manager/feed/test_opds2_serializer.py index 0455290211..9fa8ed28ee 100644 --- a/tests/manager/feed/test_opds2_serializer.py +++ b/tests/manager/feed/test_opds2_serializer.py @@ -18,11 +18,13 @@ FeedMetadata, IndirectAcquisition, Link, + LinkContentType, RichText, Series, WorkEntry, WorkEntryData, ) +from palace.manager.opds import opds2 from palace.manager.sqlalchemy.model.edition import Edition from palace.manager.sqlalchemy.model.identifier import Identifier from palace.manager.sqlalchemy.model.work import Work @@ -352,6 +354,26 @@ def test__serialize_contributor(self): assert result["sortAs"] == "Author," assert result["links"] == [{"href": "http://author", "rel": "contributor"}] + def test__serialize_contributor_resolves_link_content_type(self): + author = Author( + name="Author", + sort_name="Author,", + link=Link( + href="http://author", + rel="contributor", + type=LinkContentType.OPDS_FEED, + ), + ) + serializer = OPDS2Serializer() + result = serializer._dump_model(serializer._serialize_contributor(author)) + assert result["links"] == [ + { + "href": "http://author", + "rel": "contributor", + "type": opds2.Feed.content_type(), + } + ] + def test_serialize_opds_message(self): assert OPDS2Serializer().serialize_opds_message( OPDSMessage("URN", 200, "Description") @@ -577,6 +599,22 @@ def test_acquisition_link_type_fallback_to_indirect(self): dumped = serializer._dump_model(result) assert dumped["type"] == "application/epub+zip" + def test_acquisition_link_type_fallback_to_semantic_indirect(self): + """Semantic indirect types are resolved when direct type is missing.""" + serializer = OPDS2Serializer() + acquisition = Acquisition( + href="http://indirect-type", + rel="acquisition", + type=None, + indirect_acquisitions=[ + IndirectAcquisition(type=LinkContentType.OPDS_ENTRY), + ], + ) + result = serializer._serialize_acquisition_link(acquisition) + assert result is not None + dumped = serializer._dump_model(result) + assert dumped["type"] == opds2.BasePublication.content_type() + def test_indirect_acquisition_without_type(self): """An indirect acquisition with type=None is skipped.""" serializer = OPDS2Serializer() @@ -911,3 +949,90 @@ def test_navigation_skips_non_navigation_data_entries(self): assert len(nav) == 1 assert nav[0]["title"] == "Real Nav" assert nav[0]["href"] == "http://nav" + + def test_resolve_type_maps_opds_feed(self): + """LinkContentType.OPDS_FEED maps to OPDS2 feed content type.""" + serializer = OPDS2Serializer() + assert ( + serializer._resolve_type(LinkContentType.OPDS_FEED) + == opds2.Feed.content_type() + ) + + def test_resolve_type_maps_opds_entry(self): + """LinkContentType.OPDS_ENTRY maps to OPDS2 publication content type.""" + serializer = OPDS2Serializer() + assert ( + serializer._resolve_type(LinkContentType.OPDS_ENTRY) + == opds2.BasePublication.content_type() + ) + + def test_resolve_type_passes_through_concrete_types(self): + """Concrete content types are passed through unchanged.""" + serializer = OPDS2Serializer() + assert serializer._resolve_type("text/html") == "text/html" + assert serializer._resolve_type(None) is None + + def test_feed_link_resolves_link_content_type(self): + """Feed links with LinkContentType are resolved to OPDS2 types.""" + serializer = OPDS2Serializer() + link = Link( + href="http://example.com/shelf", + rel="http://opds-spec.org/shelf", + type=LinkContentType.OPDS_FEED, + ) + result = serializer._serialize_feed_link(link) + assert result is not None + assert result.type == opds2.Feed.content_type() + + def test_acquisition_link_resolves_link_content_type(self): + """Acquisition links with LinkContentType.OPDS_ENTRY are resolved.""" + serializer = OPDS2Serializer() + link = Acquisition( + href="http://example.com/borrow", + rel="http://opds-spec.org/acquisition/borrow", + type=LinkContentType.OPDS_ENTRY, + ) + result = serializer._acquisition_link_type(link) + assert result == opds2.BasePublication.content_type() + + def test_publication_links_resolve_link_content_type(self): + """Publication other_links with LinkContentType are resolved.""" + serializer = OPDS2Serializer() + data = WorkEntryData( + title="Test", + identifier="urn:id", + image_links=[Link(href="http://image", rel="image", type="image/png")], + acquisition_links=[ + Acquisition( + href="http://acq", + rel=OPDSFeed.OPEN_ACCESS_REL, + type="application/epub+zip", + ) + ], + other_links=[ + Link( + href="http://example.com/recommendations", + rel="recommendations", + type=LinkContentType.OPDS_FEED, + title="Recommended Works", + ), + ], + ) + publication = serializer._publication(data) + # Find the recommendations link in the publication links + rec_links = [ + link for link in publication.links if link.rel == "recommendations" + ] + assert len(rec_links) == 1 + assert rec_links[0].type == opds2.Feed.content_type() + + def test_profile_link_keeps_standard_rel(self): + """Profile link with standard 'profile' rel is kept as-is in OPDS2.""" + serializer = OPDS2Serializer() + link = Link( + href="http://example.com/profile", + rel="profile", + ) + result = serializer._serialize_feed_link(link) + assert result is not None + assert result.rel == "profile" diff --git a/tests/manager/feed/test_opds_acquisition_feed.py b/tests/manager/feed/test_opds_acquisition_feed.py index fe238a2c3e..273a2708cb 100644 --- a/tests/manager/feed/test_opds_acquisition_feed.py +++ b/tests/manager/feed/test_opds_acquisition_feed.py @@ -36,6 +36,7 @@ Acquisition, FeedData, Link, + LinkContentType, WorkEntry, WorkEntryData, ) @@ -328,7 +329,7 @@ def __call__(self, *args): # for each call to the mock method. l1, l2 = feed.links for l in l1, l2: - assert mock.attrs == l.link_attribs() + assert l.href == mock.attrs["href"] OPDSAcquisitionFeed._entrypoint_link = old_entrypoint_link # If there is only one facet in the facet group, no links are @@ -597,7 +598,7 @@ def test_format_types(self, db: DatabaseTransactionFixture): DeliveryMechanism.OVERDRIVE_DRM, ) assert [ - OPDSFeed.ENTRY_TYPE, + LinkContentType.OPDS_ENTRY, Representation.TEXT_HTML_MEDIA_TYPE + DeliveryMechanism.STREAMING_PROFILE, ] == m(overdrive_streaming_text) diff --git a/tests/manager/feed/test_opds_serializer.py b/tests/manager/feed/test_opds_serializer.py index 80ca240e54..466aea26d0 100644 --- a/tests/manager/feed/test_opds_serializer.py +++ b/tests/manager/feed/test_opds_serializer.py @@ -4,6 +4,7 @@ import pytz from lxml import etree +from palace.manager.core.user_profile import ProfileController from palace.manager.feed.serializer.opds import ( OPDS1Version1Serializer, OPDS1Version2Serializer, @@ -17,6 +18,7 @@ FeedData, IndirectAcquisition, Link, + LinkContentType, Rating, RichText, Series, @@ -447,3 +449,80 @@ def test_serialize_work_entry_with_subtitle_equals_none(self): child = element.findall(f"{{{OPDSFeed.SCHEMA_NS}}}alternativeHeadline") assert len(child) == 1 assert child[0].text == "test" + + def test_resolve_type_maps_link_content_types(self): + """LinkContentType values are mapped to OPDS1-specific content types.""" + serializer = OPDS1Version1Serializer() + assert ( + serializer._resolve_type(LinkContentType.OPDS_FEED) + == OPDSFeed.ACQUISITION_FEED_TYPE + ) + assert ( + serializer._resolve_type(LinkContentType.OPDS_ENTRY) == OPDSFeed.ENTRY_TYPE + ) + + def test_resolve_type_passes_through_concrete_types(self): + """Concrete content types are passed through unchanged.""" + serializer = OPDS1Version1Serializer() + assert serializer._resolve_type("text/html") == "text/html" + assert serializer._resolve_type("application/json") == "application/json" + assert serializer._resolve_type(None) is None + + def test_resolve_rel_maps_profile(self): + """Standard 'profile' rel is mapped to Palace-specific profile rel.""" + serializer = OPDS1Version1Serializer() + assert serializer._resolve_rel("profile") == ProfileController.LINK_RELATION + + def test_resolve_rel_passes_through_other_rels(self): + """Non-mapped rels are passed through unchanged.""" + serializer = OPDS1Version1Serializer() + assert serializer._resolve_rel("self") == "self" + assert serializer._resolve_rel("alternate") == "alternate" + assert serializer._resolve_rel(None) is None + + def test_serialize_link_resolves_link_content_type(self): + """_serialize_link resolves LinkContentType to OPDS1 type.""" + link = Link( + href="http://example.com/feed", + rel="http://opds-spec.org/shelf", + type=LinkContentType.OPDS_FEED, + ) + serializer = OPDS1Version1Serializer() + element = serializer._serialize_link(link) + assert element.get("type") == OPDSFeed.ACQUISITION_FEED_TYPE + + def test_serialize_link_resolves_profile_rel(self): + """_serialize_link resolves standard 'profile' rel to Palace-specific rel.""" + link = Link( + href="http://example.com/profile", + rel="profile", + ) + serializer = OPDS1Version1Serializer() + element = serializer._serialize_link(link) + assert element.get("rel") == ProfileController.LINK_RELATION + + def test_serialize_acquisition_link_resolves_link_content_type(self): + """_serialize_acquisition_link resolves LinkContentType to OPDS1 type.""" + link = Acquisition( + href="http://example.com/borrow", + rel="http://opds-spec.org/acquisition/borrow", + type=LinkContentType.OPDS_ENTRY, + ) + serializer = OPDS1Version1Serializer() + element = serializer._serialize_acquisition_link(link) + assert element.get("type") == OPDSFeed.ENTRY_TYPE + + def test_serialize_acquisition_link_resolves_indirect_content_type(self): + """Indirect acquisition LinkContentType values are resolved to OPDS1 types.""" + link = Acquisition( + href="http://example.com/borrow", + rel="http://opds-spec.org/acquisition/borrow", + indirect_acquisitions=[ + IndirectAcquisition(type=LinkContentType.OPDS_ENTRY) + ], + ) + serializer = OPDS1Version1Serializer() + element = serializer._serialize_acquisition_link(link) + indirect = element.find(f"{{{OPDSFeed.OPDS_NS}}}indirectAcquisition") + assert indirect is not None + assert indirect.get("type") == OPDSFeed.ENTRY_TYPE