Add PULP_MANIFEST support

repo_autoindex/_impl/api.py

@@ -1,21 +1,28 @@
 import gzip
 import logging
 from collections.abc import AsyncGenerator
-from typing import Optional
+from typing import Optional, Type
 
 import aiohttp
 
-from .base import Fetcher, GeneratedIndex
+from .base import Fetcher, GeneratedIndex, Repo
 from .yum import YumRepo
+from .pulp import PulpFileRepo
 
 LOG = logging.getLogger("repo-autoindex")
-REPO_TYPES = [YumRepo]
+REPO_TYPES: list[Type[Repo]] = [YumRepo, PulpFileRepo]
 
 
 def http_fetcher(session: aiohttp.ClientSession) -> Fetcher:
-    async def get_content_with_session(url: str) -> str:
+    async def get_content_with_session(url: str) -> Optional[str]:
         LOG.info("Fetching: %s", url)
         async with session.get(url) as resp:
+            if resp.status == 404:
+                # This error status means we successfully determined that
+                # no content exists
+                return None
+
+            # Any other error status is fatal
             resp.raise_for_status()
 
             # Deal with the non-ideal content negotiation

repo_autoindex/_impl/base.py

@@ -10,6 +10,8 @@ Fetcher = Callable[[str], Awaitable[Optional[str]]]
 
 ICON_FOLDER = "📂"
 ICON_PACKAGE = "📦"
+ICON_OPTICAL = "📀"
+ICON_QCOW = "🐮"
 ICON_OTHER = "  "
 
 

repo_autoindex/_impl/cmd.py

@@ -11,12 +11,17 @@ LOG = logging.getLogger("repo-autoindex")
 
 async def dump_autoindices(args: argparse.Namespace) -> None:
     index_filename = args.index_filename
+    wrote_any = False
     async for index in autoindex(args.url, index_href_suffix=index_filename):
         os.makedirs(index.relative_dir or ".", exist_ok=True)
         output = os.path.join(index.relative_dir or ".", index_filename)
         with open(output, "w") as f:
             f.write(index.content)
         LOG.info("Wrote %s", output)
+        wrote_any = True
+
+    if not wrote_any:
+        LOG.info("No indexable content found at %s", args.url)
 
 
 def argparser() -> argparse.ArgumentParser:

repo_autoindex/_impl/pulp.py

@@ -0,0 +1,64 @@
+from typing import Optional, Type
+from collections.abc import AsyncGenerator
+import logging
+
+from .base import Repo, GeneratedIndex, Fetcher, IndexEntry, ICON_OPTICAL, ICON_QCOW
+from .template import TemplateContext
+from .tree import treeify
+
+LOG = logging.getLogger("repo-autoindex")
+
+
+class PulpFileRepo(Repo):
+    async def render_index(
+        self, index_href_suffix: str
+    ) -> AsyncGenerator[GeneratedIndex, None]:
+        all_entries: list[IndexEntry] = [
+            IndexEntry(
+                href="PULP_MANIFEST",
+                text="PULP_MANIFEST",
+                size=str(len(self.entry_point_content)),
+            )
+        ]
+
+        # PULP_MANIFEST is a series of lines like this:
+        # rhel-workstation-7.2-snapshot-2-x86_64-boot.iso,fa687b8f847b5301b6da817fdbe612558aa69c65584ec5781f3feb0c19ff8f24,379584512
+        # rhel-workstation-7.3-rc-2-x86_64-dvd.iso,eab749310c95b4751ef9df7d7906ae0b8021c8e0dbc280c3efc8e967d5e60e71,4324327424
+        # rhel-workstation-7.3-rc-1-x86_64-dvd.iso,e165919d6977e02e493605dda6a30d2d80c3f16ee3f4c3ab946d256b815dd5db,4323278848
+        # rhel-server-7.3-rc-1-x86_64-boot.iso,f760611401fd928c2840eba85a7a80653fe2dc9dc94f3cef8ec1f3e7880d4102,427819008
+
+        for line in sorted(self.entry_point_content.splitlines()):
+            components = line.split(",")
+            if len(components) != 3:
+                LOG.warning("Ignoring bad line in PULP_MANIFEST: %s", line)
+                continue
+            entry = IndexEntry(
+                href=components[0], text=components[0], size=components[2]
+            )
+            if entry.href.endswith(".iso"):
+                entry.icon = ICON_OPTICAL
+            elif entry.href.endswith(".qcow2"):
+                entry.icon = ICON_QCOW
+            all_entries.append(entry)
+
+        ctx = TemplateContext()
+        nodes = [treeify(all_entries, index_href_suffix=index_href_suffix)]
+        while nodes:
+            node = nodes.pop()
+            yield GeneratedIndex(
+                content=ctx.render_index(index_entries=node.entries),
+                relative_dir=node.relative_dir,
+            )
+            nodes.extend(node.children)
+
+    @classmethod
+    async def probe(
+        cls: Type["PulpFileRepo"], fetcher: Fetcher, url: str
+    ) -> Optional["PulpFileRepo"]:
+        manifest_url = f"{url}/PULP_MANIFEST"
+        manifest_content = await fetcher(manifest_url)
+
+        if manifest_content is None:
+            return None
+
+        return cls(url, manifest_content, fetcher)

tests/sample_pulp_repo/PULP_MANIFEST

@@ -0,0 +1,5 @@
+file1.iso,abc123,100
+file2.qcow2,abc123,200
+oops, this line ain't so valid
+
+file3,abc234,300

tests/test_cmd.py

@@ -1,5 +1,7 @@
 import pathlib
 import asyncio
+import logging
+from collections.abc import Callable, Awaitable
 
 import pytest
 
@@ -11,31 +13,45 @@ from repo_autoindex._impl.cmd import entrypoint
 THIS_DIR = pathlib.Path(__file__).parent
 
 
-async def test_command(monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path):
-    """Run the repo-autoindex command against a sample repo and check the generated index."""
-
-    monkeypatch.chdir(tmp_path)
+class CommandTester:
+    def __init__(self, monkeypatch: pytest.MonkeyPatch):
+        self.monkeypatch = monkeypatch
 
+    async def __call__(self, url: str):
         entrypoint_coro = []
 
         def fake_run(coro):
             assert asyncio.iscoroutine(coro)
             entrypoint_coro.append(coro)
 
-    monkeypatch.setattr("asyncio.run", fake_run)
+        self.monkeypatch.setattr("asyncio.run", fake_run)
 
         app = web.Application()
         app.add_routes([web.static("/", THIS_DIR)])
 
         async with test_utils.TestServer(app) as server:
-        repo_url = server.make_url("/sample_repo///")
-        monkeypatch.setattr("sys.argv", ["repo-autoindex", str(repo_url)])
+            repo_url = server.make_url(url + "//")
+            self.monkeypatch.setattr("sys.argv", ["repo-autoindex", str(repo_url)])
 
             entrypoint()
 
             assert entrypoint_coro
             await entrypoint_coro[0]
 
+
+@pytest.fixture
+def tester(monkeypatch: pytest.MonkeyPatch) -> CommandTester:
+    return CommandTester(monkeypatch)
+
+
+async def test_command_yum(
+    monkeypatch: pytest.MonkeyPatch, tester: CommandTester, tmp_path: pathlib.Path
+):
+    """Run the repo-autoindex command against a sample yum repo and check the generated index."""
+    monkeypatch.chdir(tmp_path)
+
+    await tester("/sample_repo")
+
     # It should have written index files reproducing the structure
     index_toplevel = tmp_path.joinpath("index.html")
     index_pkgs = tmp_path.joinpath("pkgs", "index.html")
@@ -56,3 +72,43 @@ async def test_command(monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path):
     assert '<a href="w/index.html">w/</a>' in pkgs
 
     assert '<a href="walrus-5.21-1.noarch.rpm">walrus-5.21-1.noarch.rpm</a>' in w
+
+
+async def test_command_pulp(
+    monkeypatch: pytest.MonkeyPatch, tester: CommandTester, tmp_path: pathlib.Path
+):
+    """Run the repo-autoindex command against a sample pulp file repo and check the generated index."""
+    monkeypatch.chdir(tmp_path)
+
+    await tester("/sample_pulp_repo")
+
+    # It should have written an index file
+    index_toplevel = tmp_path.joinpath("index.html")
+    assert index_toplevel.exists()
+
+    # Simple sanity check of some expected content
+    toplevel = index_toplevel.read_text()
+
+    assert '<a href="file2.qcow2">file2.qcow2</a>' in toplevel
+
+
+async def test_command_no_content(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: pathlib.Path,
+    tester: CommandTester,
+    caplog: pytest.LogCaptureFixture,
+):
+    """Run the repo-autoindex command against an empty repo and verify nothing is written."""
+
+    caplog.set_level(logging.INFO)
+
+    monkeypatch.chdir(tmp_path)
+
+    await tester("/some-nonexistent-dir")
+
+    # It should not have written any index.html
+    index_toplevel = tmp_path.joinpath("index.html")
+    assert not index_toplevel.exists()
+
+    # It should have mentioned that there was no content
+    assert "No indexable content found" in caplog.text

tests/test_http_fetcher.py

@@ -16,6 +16,7 @@ class FakeResponse:
     def __init__(self, body: bytes, content_type: str):
         self.body = body
         self.content_type = content_type
+        self.status = 200
 
     async def __aenter__(self):
         return self

tests/test_yum_render_typical.py

@@ -1,6 +1,5 @@
 from typing import Optional
 import textwrap
-import pytest
 
 from repo_autoindex import autoindex
 from repo_autoindex._impl.base import GeneratedIndex
@@ -395,7 +394,7 @@ class StaticFetcher:
         self.content: dict[str, str] = {}
 
     async def __call__(self, url: str) -> Optional[str]:
-        return self.content[url]
+        return self.content.get(url)
 
 
 async def test_typical_index():