rpmrepo: new RPM repo source

docs/usage.rst

@@ -857,6 +857,29 @@ strip_release
 
 Note that either pkg or srcpkg needs to be specified (but not both) or the item name will be used as pkg.
 
+Check RPM repository
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+::
+
+  source = "rpmrepo"
+
+This enables you to check latest package versions in an arbitrary RPM repository in `repomd` format used by package managers such as ``dnf`` (Fedora, RHEL, AlmaLinux etc.) or ``zypper`` (openSUSE) without the need for native RPM tools.
+
+pkg
+  Name of the RPM package (you can also use ``rpmrepo`` as with other sources, but ``pkg`` is preferred for clarity)
+
+repo
+  URL of the repository (required, ``repodata/repomd.xml`` should be there)
+
+arch
+  Architecture of the RPM package (``binary``, ``src``, ``any``, ``x86_64``, ``aarch64``, etc, defaults to ``binary``)
+
+This source supports :ref:`list options`.
+
+.. note::
+   An additional dependency "lxml" is required.
+   You can use ``pip install 'nvchecker[rpmrepo]'``.
+
 Check Git repository
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 ::

nvchecker_source/rpmrepo.py

@@ -0,0 +1,83 @@
+# MIT licensed
+# Copyright (c) 2024 Jakub Ružička <jru@debian.org>, et al.
+
+import asyncio
+import gzip
+import pathlib
+import urllib
+
+import lxml.etree
+from nvchecker.api import session, AsyncCache, Entry, KeyManager, VersionResult
+
+
+# XML namespaces used in repodata (dead links haha)
+NS = {
+    'common': 'http://linux.duke.edu/metadata/common',
+    'repo':   'http://linux.duke.edu/metadata/repo',
+    'rpm':    'http://linux.duke.edu/metadata/rpm'
+}
+
+
+async def get_version(
+  name: str, conf: Entry, *,
+  cache: AsyncCache, keymanager: KeyManager,
+  **kwargs,
+) -> VersionResult:
+  repo = conf['repo']
+  arch = conf.get('arch', 'binary')
+  pkg = conf.get('pkg')
+  if not pkg:
+    pkg = conf.get('rpmrepo', name)
+
+  repo_url = urllib.parse.urlparse(repo)
+  repo_path = pathlib.PurePosixPath(repo_url.path)
+
+  # get the url of repomd.xml
+  repomd_path = repo_path / 'repodata' / 'repomd.xml'
+  repomd_url = repo_url._replace(path=str(repomd_path)).geturl()
+  # download repomd.xml (use cache)
+  repomd_body = await cache.get(repomd_url, get_file)
+  # parse repomd.xml
+  repomd_xml = lxml.etree.fromstring(repomd_body)
+
+  # get the url of *primary.xml.gz
+  primary_element = repomd_xml.find('repo:data[@type="primary"]/repo:location', namespaces=NS)
+  primary_path = repo_path / primary_element.get('href')
+  primary_url = repo_url._replace(path=str(primary_path)).geturl()
+  # download and decompress *primary.xml.gz (use cache)
+  primary_body = await cache.get(primary_url, get_file_gz)
+  # parse *primary.xml metadata
+  metadata = lxml.etree.fromstring(primary_body)
+
+  # use set to eliminate duplication
+  versions_set = set()
+  # iterate package metadata
+  for el in metadata.findall(f'common:package[common:name="{pkg}"]', namespaces=NS):
+    pkg_arch = el.findtext('common:arch', namespaces=NS)
+
+    # filter bych arch
+    if arch == 'binary':
+      if pkg_arch == 'src':
+        continue
+    elif arch != 'any':
+      if pkg_arch != arch:
+        continue
+
+    version_info = el.find('common:version', namespaces=NS)
+    version = version_info.get('ver')
+    versions_set.add(version)
+
+  versions = list(versions_set)
+  return versions
+
+
+async def get_file(url: str) -> str:
+  res = await session.get(url)
+  return res.body
+
+
+async def get_file_gz(url: str) -> str:
+  res = await session.get(url)
+  loop = asyncio.get_running_loop()
+  return await loop.run_in_executor(
+    None, gzip.decompress, res.body)

setup.cfg

@@ -64,6 +64,8 @@ pypi =
   packaging
 htmlparser =
   lxml
+rpmrepo =
+  lxml
 jq =
   jq
 

tests/test_rpmrepo.py

@@ -0,0 +1,19 @@
+# MIT licensed
+# Copyright (c) 2024 Jakub Ružička <jru@debian.org>, et al.
+
+import pytest
+pytestmark = [pytest.mark.asyncio, pytest.mark.needs_net]
+
+async def test_rpmrepo_fedora(get_version):
+    assert await get_version("knot_fedora-39", {
+        "source": "rpmrepo",
+        "pkg": "knot",
+        "repo": "http://ftp.sh.cvut.cz/fedora/linux/updates/39/Everything/x86_64/",
+    }) == "3.3.9"
+
+async def test_rpmrepo_alma(get_version):
+    assert await get_version("knot_fedora-39", {
+        "source": "rpmrepo",
+        "pkg": "tmux",
+        "repo": "http://ftp.sh.cvut.cz/almalinux/9.5/BaseOS/x86_64/os/",
+    }) == "3.2a"