Test redirects

tests/test_feed.py

@@ -0,0 +1,177 @@
+import asyncio
+import dataclasses
+import http.server
+import threading
+import typing
+
+import requests
+
+from cry import feed
+
+
+@dataclasses.dataclass
+class MockResponse:
+    code: int = 200
+    headers: list[typing.Tuple[str, str]] | None = None
+    content: bytes = b""
+
+    def write_to(self, handler: http.server.BaseHTTPRequestHandler):
+        handler.send_response(self.code)
+        if self.headers is not None:
+            for name, value in self.headers:
+                handler.send_header(name, value)
+        handler.send_header("content-length", str(len(self.content)))
+        handler.end_headers()
+        handler.wfile.write(self.content)
+
+
+class TestWebServer:
+    __test__ = False
+    server_port: int
+    http_server: http.server.ThreadingHTTPServer
+    server_thread: threading.Thread
+
+    handlers: dict[str, list[MockResponse]]
+
+    def __init__(self):
+        server = self
+
+        class TestWebHandler(http.server.BaseHTTPRequestHandler):
+            def do_GET(self):
+                server._handle_GET(self)
+
+        self.http_server = http.server.ThreadingHTTPServer(("", 0), TestWebHandler)
+        self.server_thread = threading.Thread(target=self.http_server.serve_forever)
+        self.server_thread.daemon = True
+        self.handlers = {}
+
+        self.server_port = self.http_server.server_port
+
+    def make_url(self, path: str) -> str:
+        return f"http://localhost:{self.server_port}{path}"
+
+    def handle(
+        self,
+        path: str,
+        content: bytes = b"",
+        code: int = 200,
+        content_type: str | None = None,
+        headers: list[typing.Tuple[str, str]] | None = None,
+    ):
+        if headers is None:
+            headers = []
+        else:
+            headers = list(headers)
+
+        if content_type is not None:
+            headers.append(("content-type", content_type))
+
+        self.respond(path, MockResponse(code, headers, content))
+
+    def respond(self, path: str, response: MockResponse):
+        response_list = self.handlers.get(path)
+        if response_list is None:
+            response_list = []
+            self.handlers[path] = response_list
+
+        response_list.append(response)
+
+    def _handle_GET(self, handler: http.server.BaseHTTPRequestHandler):
+        responses = self.handlers.get(handler.path, [])
+
+        if len(responses) > 0:
+            response = responses[0]
+            if len(responses) > 1:
+                responses.pop(0)
+        else:
+            response = MockResponse(code=404)
+
+        response.write_to(handler)
+
+    def __enter__(self):
+        self.server_thread.start()
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        del exc_type
+        del exc_value
+        del traceback
+        self.http_server.shutdown()
+
+
+TEST_FEED = b"""
+<?xml version="1.0" encoding="utf-8" standalone="yes" ?>
+<rss version="2.0">
+  <channel>
+    <title>Smallest</title>
+    <item>
+      <title>An Item</title>
+      <link>http://example.com</link>
+      <guid>tag:d0ty.me,2024-01-01:smallest:one</guid>
+    </item>
+  </channel>
+</rss>
+"""
+
+
+def test_basic_successful_fetch():
+    with TestWebServer() as server:
+        server.handle("/", TEST_FEED, content_type="text/xml")
+
+        meta = feed.FeedMeta.from_url(server.make_url("/"), "asdf")
+        result, new_meta = asyncio.run(feed.fetch_feed(meta))
+
+        assert new_meta.url == meta.url
+        assert isinstance(result, feed.Feed)
+        assert len(result.entries) == 1
+        assert result.title == "Smallest"
+
+
+def test_fetch_after_temp_redirect():
+    with TestWebServer() as server:
+        server.handle("/old", code=307, headers=[("location", "/temp")])
+        server.handle("/temp", TEST_FEED, content_type="text/xml")
+
+        meta = feed.FeedMeta.from_url(server.make_url("/old"), "asdf")
+        result, new_meta = asyncio.run(feed.fetch_feed(meta))
+        assert new_meta.url == meta.url
+        assert isinstance(result, feed.Feed)
+
+
+def test_fetch_after_permanent_redirect():
+    with TestWebServer() as server:
+        server.handle("/old", code=308, headers=[("location", "/perm")])
+        server.handle("/perm", TEST_FEED, content_type="text/xml")
+
+        meta = feed.FeedMeta.from_url(server.make_url("/old"), "asdf")
+        result, new_meta = asyncio.run(feed.fetch_feed(meta))
+        assert new_meta.url == server.make_url("/perm")
+        assert isinstance(result, feed.Feed)
+
+
+def test_fetch_after_permanent_to_temporary_redirect():
+    with TestWebServer() as server:
+        server.handle("/old", code=308, headers=[("location", "/perm")])
+        server.handle("/perm", code=307, headers=[("location", "/temp")])
+        server.handle("/temp", TEST_FEED, content_type="text/xml")
+
+        meta = feed.FeedMeta.from_url(server.make_url("/old"), "asdf")
+        result, new_meta = asyncio.run(feed.fetch_feed(meta))
+
+        # NOTE: we should record the PERMANENT redirect, not the temporary one.
+        assert new_meta.url == server.make_url("/perm")
+        assert isinstance(result, feed.Feed)
+
+
+def test_fetch_after_permanent_to_permanent_redirect():
+    with TestWebServer() as server:
+        server.handle("/old", code=308, headers=[("location", "/one")])
+        server.handle("/one", code=308, headers=[("location", "/two")])
+        server.handle("/two", TEST_FEED, content_type="text/xml")
+
+        meta = feed.FeedMeta.from_url(server.make_url("/old"), "asdf")
+        result, new_meta = asyncio.run(feed.fetch_feed(meta))
+
+        # NOTE: we should record the latest redirect.
+        assert new_meta.url == server.make_url("/two")
+        assert isinstance(result, feed.Feed)