Fix throwing error with the push notification for deleted video

SeoulSKY · Sep 10, 2024 · 74f4f3a · 74f4f3a
1 parent 1adfbc5
commit 74f4f3a
Show file tree

Hide file tree

Showing 2 changed files with 32 additions and 9 deletions.
diff --git a/tests/test_youtube_notifier.py b/tests/test_youtube_notifier.py
@@ -15,6 +15,8 @@
     "UCupvZG-5ko_eiXAupbDfxWw",
 ]
 
+# ruff: noqa: E501 ERA001
+
 xmls = [
     """
     <feed xmlns:yt="http://www.youtube.com/xml/schemas/2015" xmlns="http://www.w3.org/2005/Atom">
@@ -95,8 +97,22 @@
       </entry>
     </feed>
     """,
+    """
+    <feed xmlns:at="http://purl.org/atompub/tombstones/1.0" xmlns="http://www.w3.org/2005/Atom">
+        <at:deleted-entry ref="yt:video:VIDEO_ID" when="2024-09-09T22:34:19.642702+00:00">
+            <link href="https://www.youtube.com/watch?v=VIDEO_ID" />
+            <at:by>
+                <name>Channel title</name>
+                <uri>https://www.youtube.com/channel/CHANNEL_ID</uri>
+            </at:by>
+        </at:deleted-entry>
+    </feed>
+    """,
 ]
 
+# ruff: enable
+
+
 @pytest.fixture(scope="session")
 def notifier() -> YouTubeNotifier:
     """Setup/Teardown code that runs before and after the tests in this package."""
@@ -279,7 +295,7 @@ def test_post(notifier: YouTubeNotifier) -> None:
     assert response.status_code == HTTPStatus.BAD_REQUEST
 
     with pytest.raises(RuntimeError):
-        response = client.post(CALLBACK_URL, headers=headers, content="<feed/>")
+        client.post(CALLBACK_URL, headers=headers, content="<feed/>")
 
     password = notifier._config.password
     notifier._config.password = "password"  # noqa: S105

diff --git a/ytnoti/__init__.py b/ytnoti/__init__.py
@@ -558,7 +558,7 @@ async def _verify_channel_ids(channel_ids: Iterable[str]) -> None:
                 response = await client.head(
                     f"https://www.youtube.com/feeds/videos.xml?channel_id={channel_id}"
                 )
-                if response.status_code != HTTPStatus.OK.value:
+                if response.status_code != HTTPStatus.OK:
                     raise ValueError(f"Invalid channel ID: {channel_id}")
 
 
@@ -654,7 +654,7 @@ async def _register(
                     headers={"Content-type": "application/x-www-form-urlencoded"},
                 )
 
-            if response.status_code == HTTPStatus.CONFLICT.value:  # pragma: no cover
+            if response.status_code == HTTPStatus.CONFLICT:  # pragma: no cover
                 if not self.is_ready:
                     raise ConnectionError(
                         f"Cannot {mode} while the server is not ready"
@@ -667,7 +667,7 @@ async def _register(
                     response.status_code,
                 )
 
-            if response.status_code != HTTPStatus.NO_CONTENT.value:  # pragma: no cover
+            if response.status_code != HTTPStatus.NO_CONTENT:  # pragma: no cover
                 raise HTTPError(
                     f"Failed to {mode} channel: {channel_id}", response.status_code
                 )
@@ -700,23 +700,30 @@ async def _get(request: Request) -> Response:
         """Handle a challenge from the Google pubsubhubbub server."""
         challenge = request.query_params.get("hub.challenge")
         if challenge is None:
-            return Response(status_code=HTTPStatus.BAD_REQUEST.value)
+            return Response(status_code=HTTPStatus.BAD_REQUEST)
 
         return Response(challenge)
 
     async def _post(self, request: Request) -> Response:
         """Handle push notifications from the Google pubsubhubbub server."""
         if not await self._is_authorized(request):
-            return Response(status_code=HTTPStatus.UNAUTHORIZED.value)
+            return Response(status_code=HTTPStatus.UNAUTHORIZED)
+
+        body = await request.body()
 
         try:
-            body = xmltodict.parse(await request.body())
+            body = xmltodict.parse(body)
         except ExpatError:
-            return Response(status_code=HTTPStatus.BAD_REQUEST.value)
+            self._logger.debug("Received invalid request body: %s", body)
+            return Response(status_code=HTTPStatus.BAD_REQUEST)
 
         self._logger.debug("Received push notification: %s", body)
 
         try:
+            if "at:deleted-entry" in body["feed"]:
+                self._logger.debug("Ignoring push notification for deleted video")
+                return Response(status_code=HTTPStatus.NO_CONTENT)
+
             # entry can be list of dict or just dict
             entries = (
                 body["feed"]["entry"]
@@ -766,7 +773,7 @@ async def _post(self, request: Request) -> Response:
         except (TypeError, KeyError, ValueError) as ex:
             raise RuntimeError(f"Failed to parse request body: {body}") from ex
 
-        return Response(status_code=HTTPStatus.NO_CONTENT.value)
+        return Response(status_code=HTTPStatus.NO_CONTENT)
 
     @staticmethod
     def _parse_timestamp(timestamp: str) -> datetime: