PP-490 Fixed the lcp and adobe drm tag names in the annotators

core/feed/acquisition.py

@@ -74,9 +74,9 @@ def __init__(
     def generate_feed(self, annotate: bool = True) -> None:
         """Generate the feed metadata and links.
         We assume the entries have already been annotated."""
-        self._feed.add_metadata("id", text=self.url)
-        self._feed.add_metadata("title", text=self.title)
-        self._feed.add_metadata("updated", text=strftime(utc_now()))
+        self._feed.metadata.id = self.url
+        self._feed.metadata.title = self.title
+        self._feed.metadata.updated = strftime(utc_now())
         self._feed.add_link(href=self.url, rel="self")
         if annotate:
             self.annotator.annotate_feed(self._feed)

core/feed/annotator/circulation.py

@@ -1503,7 +1503,7 @@ def adobe_id_tags(
                 drm_licensor = FeedEntryType.create(
                     vendor=vendor_id, clientToken=FeedEntryType(text=token)
                 )
-                cached = {"licensor": drm_licensor}
+                cached = {"drm_licensor": drm_licensor}
 
             self._adobe_id_cache[cache_key] = cached
         else:
@@ -1525,7 +1525,9 @@ def lcp_key_retrieval_tags(self, active_loan: Loan) -> Dict[str, FeedEntryType]:
             hashed_passphrase: LCPHashedPassphrase = (
                 lcp_credential_factory.get_hashed_passphrase(db, active_loan.patron)
             )
-            response["hashed_passphrase"] = FeedEntryType(text=hashed_passphrase.hashed)
+            response["lcp_hashed_passphrase"] = FeedEntryType(
+                text=hashed_passphrase.hashed
+            )
         except LCPError:
             # The patron's passphrase wasn't generated yet and not present in the database.
             pass
@@ -1544,7 +1546,7 @@ def add_patron(self, feed: FeedData) -> None:
             ] = self.patron.authorization_identifier
 
         patron_tag = FeedEntryType.create(**patron_details)
-        feed.add_metadata("patron", patron_tag)
+        feed.metadata.patron = patron_tag
 
     def add_authentication_document_link(self, feed_obj: FeedData) -> None:
         """Create a <link> tag that points to the circulation

core/feed/annotator/loan_and_hold.py

@@ -87,8 +87,8 @@ def annotate_feed(self, feed: FeedData) -> None:
             link = self.user_profile_management_protocol_link
             if link.href is not None:
                 feed.add_link(link.href, rel=link.rel)
-            for name, value in tags.items():
-                feed.add_metadata(name, feed_entry=value)
+            if "drm_licensor" in tags:
+                feed.metadata.drm_licensor = tags["drm_licensor"]
 
     def annotate_work_entry(
         self, entry: WorkEntry, updated: Optional[datetime] = None

core/feed/navigation.py

@@ -51,9 +51,9 @@ def navigation(
         return feed
 
     def generate_feed(self) -> None:
-        self._feed.add_metadata("title", text=self.title)
-        self._feed.add_metadata("id", text=self.url)
-        self._feed.add_metadata("updated", text=strftime(utc_now()))
+        self._feed.metadata.title = self.title
+        self._feed.metadata.id = self.url
+        self._feed.metadata.updated = strftime(utc_now())
         self._feed.add_link(href=self.url, rel="self")
         if not self.lane.children:
             # We can't generate links to children, since this Worklist

core/feed/serializer/opds.py

@@ -13,6 +13,7 @@
     DataEntry,
     FeedData,
     FeedEntryType,
+    FeedMetadata,
     IndirectAcquisition,
     WorkEntryData,
 )
@@ -78,9 +79,7 @@
         if feed.entrypoint:
             serialized.set(f"{{{OPDSFeed.SIMPLIFIED_NS}}}entrypoint", feed.entrypoint)
 
-        for name, metadata in feed.metadata.items():
-            element = self._serialize_feed_entry(name, metadata)
-            serialized.append(element)
+        serialized.extend(self._serialize_feed_metadata(feed.metadata))
 
         for entry in feed.entries:
             if entry.computed:
@@ -113,6 +112,28 @@
         etree.indent(serialized)
         return self.to_string(serialized)
 
+    def _serialize_feed_metadata(self, metadata: FeedMetadata) -> List[etree._Element]:
+        tags = []
+        # Compulsory title
+        tags.append(self._tag("title", metadata.title or ""))
+
+        if metadata.id:
+            tags.append(self._tag("id", metadata.id))
+        if metadata.updated:
+            tags.append(self._tag("updated", metadata.updated))
+        if metadata.patron:
+            tags.append(self._serialize_feed_entry("patron", metadata.patron))
+        if metadata.drm_licensor:
+            tags.append(self._serialize_feed_entry("licensor", metadata.drm_licensor))
+        if metadata.lcp_hashed_passphrase:
+            tags.append(
+                self._serialize_feed_entry(
+                    "hashed_passphrase", metadata.lcp_hashed_passphrase
+                )
+            )
+
+        return tags
+
     def serialize_work_entry(self, feed_entry: WorkEntryData) -> etree._Element:
         entry: etree._Element = OPDSFeed.entry()
 

core/feed/serializer/opds2.py

@@ -54,10 +54,10 @@ def serialize_feed(
     def _serialize_metadata(self, feed: FeedData) -> Dict[str, Any]:
         fmeta = feed.metadata
         metadata: Dict[str, Any] = {}
-        if title := fmeta.get("title"):
-            metadata["title"] = title.text
-        if item_count := fmeta.get("items_per_page"):
-            metadata["itemsPerPage"] = int(item_count.text or 0)
+        if fmeta.title:
+            metadata["title"] = fmeta.title
+        if fmeta.items_per_page is not None:
+            metadata["itemsPerPage"] = fmeta.items_per_page
         return metadata
 
     def serialize_opds_message(self, entry: OPDSMessage) -> Dict[str, Any]:

core/feed/types.py

@@ -201,6 +201,17 @@ def __init__(
         self.license_pool = license_pool
 
 
+@dataclass
+class FeedMetadata(BaseModel):
+    title: Optional[str] = None
+    id: Optional[str] = None
+    updated: Optional[str] = None
+    items_per_page: Optional[int] = None
+    patron: Optional[FeedEntryType] = None
+    drm_licensor: Optional[FeedEntryType] = None
+    lcp_hashed_passphrase: Optional[FeedEntryType] = None
+
+
 class DataEntryTypes:
     NAVIGATION = "navigation"
 
@@ -222,19 +233,11 @@ class FeedData(BaseModel):
     facet_links: List[Link] = field(default_factory=list)
     entries: List[WorkEntry] = field(default_factory=list)
     data_entries: List[DataEntry] = field(default_factory=list)
-    metadata: Dict[str, FeedEntryType] = field(default_factory=dict)
+    metadata: FeedMetadata = field(default_factory=lambda: FeedMetadata())
     entrypoint: Optional[str] = None
 
     class Config:
         arbitrary_types_allowed = True
 
     def add_link(self, href: str, **kwargs: Any) -> None:
         self.links.append(Link(href=href, **kwargs))
-
-    def add_metadata(
-        self, name: str, feed_entry: Optional[FeedEntryType] = None, **kwargs: Any
-    ) -> None:
-        if not feed_entry:
-            self.metadata[name] = FeedEntryType(**kwargs)
-        else:
-            self.metadata[name] = feed_entry

tests/api/feed/test_library_annotator.py

@@ -262,8 +262,7 @@ def test_fulfill_link_includes_device_registration_tags(
             pool, loan, other_delivery_mechanism
         )
         assert link is not None
-        for name, child in link:
-            assert name != "licensor"
+        assert link.drm_licensor is None
 
         # No new Credential has been associated with the patron.
         assert old_credentials == patron.credentials
@@ -273,8 +272,8 @@ def test_fulfill_link_includes_device_registration_tags(
         link = annotator_fixture.annotator.fulfill_link(
             pool, loan, adobe_delivery_mechanism
         )
-        licensor = getattr(link, "licensor", None)
-        assert None != licensor
+        assert link is not None
+        assert link.drm_licensor is not None
 
         # An Adobe ID-specific identifier has been created for the patron.
         [adobe_id_identifier] = [
@@ -293,7 +292,8 @@ def test_fulfill_link_includes_device_registration_tags(
         expect = annotator_fixture.annotator.adobe_id_tags(
             adobe_id_identifier.credential
         )
-        assert expect.get("licensor") == licensor
+        assert link is not None
+        assert expect.get("drm_licensor") == link.drm_licensor
 
     def test_no_adobe_id_tags_when_vendor_id_not_configured(
         self, annotator_fixture: LibraryAnnotatorFixture
@@ -318,12 +318,12 @@ def test_adobe_id_tags_when_vendor_id_configured(
         patron_identifier = "patron identifier"
         element = annotator_fixture.annotator.adobe_id_tags(patron_identifier)
 
-        assert "licensor" in element
+        assert "drm_licensor" in element
         assert vendor_id_fixture.TEST_VENDOR_ID == getattr(
-            element["licensor"], "vendor", None
+            element["drm_licensor"], "vendor", None
         )
 
-        token = getattr(element["licensor"], "clientToken", None)
+        token = getattr(element["drm_licensor"], "clientToken", None)
         assert token is not None
         # token.text is a token which we can decode, since we know
         # the secret.
@@ -338,7 +338,7 @@ def test_adobe_id_tags_when_vendor_id_configured(
         # object that renders to the same data.
         same_tag = annotator_fixture.annotator.adobe_id_tags(patron_identifier)
         assert same_tag is not element
-        assert same_tag["licensor"].dict() == element["licensor"].dict()
+        assert same_tag["drm_licensor"].dict() == element["drm_licensor"].dict()
 
         # If the Adobe Vendor ID configuration is present but
         # incomplete, adobe_id_tags does nothing.
@@ -375,15 +375,16 @@ def test_lcp_acquisition_link_contains_hashed_passphrase(
         link = annotator_fixture.annotator.fulfill_link(
             pool, loan, other_delivery_mechanism
         )
-        assert not hasattr(link, "hashed_passphrase")
+        assert link is not None
+        assert link.lcp_hashed_passphrase is None
 
         # The fulfill link for lcp DRM includes hashed_passphrase
         link = annotator_fixture.annotator.fulfill_link(
             pool, loan, lcp_delivery_mechanism
         )
-        hashed_passphrase = getattr(link, "hashed_passphrase", None)
-        assert hashed_passphrase is not None
-        assert hashed_passphrase.text == hashed_password.hashed
+        assert link is not None
+        assert link.lcp_hashed_passphrase is not None
+        assert link.lcp_hashed_passphrase.text == hashed_password.hashed
 
     def test_default_lane_url(self, annotator_fixture: LibraryAnnotatorFixture):
         default_lane_url = annotator_fixture.annotator.default_lane_url()
@@ -1411,14 +1412,14 @@ def test_drm_device_registration_feed_tags(
         # The feed-level tag has the drm:scheme attribute set.
         assert (
             "http://librarysimplified.org/terms/drm/scheme/ACS"
-            == feed_tag["licensor"].scheme
+            == feed_tag["drm_licensor"].scheme
         )
 
         # If we remove that attribute, the feed-level tag is the same as the
         # generic tag.
-        assert feed_tag["licensor"].dict() != generic_tag["licensor"].dict()
-        delattr(feed_tag["licensor"], "scheme")
-        assert feed_tag["licensor"].dict() == generic_tag["licensor"].dict()
+        assert feed_tag["drm_licensor"].dict() != generic_tag["drm_licensor"].dict()
+        delattr(feed_tag["drm_licensor"], "scheme")
+        assert feed_tag["drm_licensor"].dict() == generic_tag["drm_licensor"].dict()
 
     def test_borrow_link_raises_unfulfillable_work(
         self, annotator_fixture: LibraryAnnotatorFixture

tests/api/feed/test_opds2_serializer.py

@@ -6,6 +6,7 @@
     Author,
     FeedData,
     FeedEntryType,
+    FeedMetadata,
     IndirectAcquisition,
     Link,
     WorkEntry,
@@ -20,9 +21,9 @@
 class TestOPDS2Serializer:
     def test_serialize_feed(self):
         feed = FeedData(
-            metadata=dict(
-                items_per_page=FeedEntryType(text="20"),
-                title=FeedEntryType(text="Title"),
+            metadata=FeedMetadata(
+                title="Title",
+                items_per_page=20,
             )
         )
         w = WorkEntry(

tests/api/feed/test_opds_acquisition_feed.py

@@ -1402,11 +1402,11 @@ def test_navigation_with_sublanes(
 
         feed = response._feed
 
-        assert "Navigation" == feed.metadata["title"].text
+        assert "Navigation" == feed.metadata.title
         [self_link] = feed.links
         assert "http://navigation" == self_link.href
         assert "self" == self_link.rel
-        assert "http://navigation" == feed.metadata["id"].text
+        assert "http://navigation" == feed.metadata.id
         [fantasy, romance] = sorted(feed.data_entries, key=lambda x: x.title or "")
 
         assert data.fantasy.display_name == fantasy.title
@@ -1436,11 +1436,11 @@ def test_navigation_without_sublanes(
             session, "Navigation", "http://navigation", data.fantasy, MockAnnotator()
         )
         parsed = feed._feed
-        assert "Navigation" == parsed.metadata["title"].text
+        assert "Navigation" == parsed.metadata.title
         [self_link] = parsed.links
         assert "http://navigation" == self_link.href
         assert "self" == self_link.rel
-        assert "http://navigation" == parsed.metadata["id"].text
+        assert "http://navigation" == parsed.metadata.id
         [fantasy] = parsed.data_entries
 
         assert "All " + data.fantasy.display_name == fantasy.title

tests/api/feed/test_opds_serializer.py

@@ -96,20 +96,37 @@ def test__serialize_acquistion_link(self):
             copies_total="1",
             availability_status="available",
             indirect_acquisitions=[IndirectAcquisition(type="indirect")],
+            lcp_hashed_passphrase=FeedEntryType(text="passphrase"),
+            drm_licensor=FeedEntryType.create(
+                vendor="vendor", clientToken=FeedEntryType(text="token")
+            ),
         )
         element = OPDS1Serializer()._serialize_acquistion_link(link)
         assert element.tag == "link"
         assert dict(element.attrib) == dict(href=link.href)
 
-        for child in element:
-            if child.tag == f"{{{OPDSFeed.OPDS_NS}}}indirectAcquisition":
-                assert child.get("type") == "indirect"
-            elif child.tag == f"{{{OPDSFeed.OPDS_NS}}}holds":
-                assert child.get("total") == "0"
-            elif child.tag == f"{{{OPDSFeed.OPDS_NS}}}copies":
-                assert child.get("total") == "1"
-            elif child.tag == f"{{{OPDSFeed.OPDS_NS}}}availability":
-                assert child.get("status") == "available"
+        tests = [
+            (
+                f"{{{OPDSFeed.OPDS_NS}}}indirectAcquisition",
+                lambda child: child.get("type") == "indirect",
+            ),
+            (f"{{{OPDSFeed.OPDS_NS}}}holds", lambda child: child.get("total") == "0"),
+            (f"{{{OPDSFeed.OPDS_NS}}}copies", lambda child: child.get("total") == "1"),
+            (
+                f"{{{OPDSFeed.OPDS_NS}}}availability",
+                lambda child: child.get("status") == "available",
+            ),
+            ("hashed_passphrase", lambda child: child.text == "passphrase"),
+            (
+                f"{{{OPDSFeed.DRM_NS}}}licensor",
+                lambda child: child.get(f"{{{OPDSFeed.DRM_NS}}}vendor") == "vendor"
+                and child[0].text == "token",
+            ),
+        ]
+        for tag, test_fn in tests:
+            children = element.findall(tag)
+            assert len(children) == 1
+            assert test_fn(children[0])
 
     def test_serialize_work_entry(self):
         data = WorkEntryData(