Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

rpmrepo: new RPM repo source #282

Merged
merged 2 commits into from
Dec 2, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 24 additions & 1 deletion docs/usage.rst
Original file line number Diff line number Diff line change
Expand Up @@ -327,7 +327,7 @@ post_data_type
You can use ``pip install 'nvchecker[htmlparser]'``.

Search with an JSON Parser (jq)
~~~~~~~~~~~~~~~~~~~~~~~~~~
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
::

source = "jq"
Expand Down Expand Up @@ -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
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
::
Expand Down
83 changes: 83 additions & 0 deletions nvchecker_source/rpmrepo.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,83 @@
# MIT licensed
# Copyright (c) 2024 Jakub Ružička <[email protected]>, 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)
2 changes: 2 additions & 0 deletions setup.cfg
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,8 @@ pypi =
packaging
htmlparser =
lxml
rpmrepo =
lxml
jq =
jq

Expand Down
19 changes: 19 additions & 0 deletions tests/test_rpmrepo.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,19 @@
# MIT licensed
# Copyright (c) 2024 Jakub Ružička <[email protected]>, 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"
Loading