From 5622d6d0c1bc97ea8b51283d84ee9843d2f5d610 Mon Sep 17 00:00:00 2001
From: DDoSolitary <DDoSolitary@gmail.com>
Date: Thu, 5 Nov 2020 02:26:25 +0800
Subject: [PATCH] add ALPM source

---
 docs/usage.rst           |  23 ++++++++
 nvchecker_source/alpm.py |  34 ++++++++++++
 tests/test_alpm.py       | 112 +++++++++++++++++++++++++++++++++++++++
 3 files changed, 169 insertions(+)
 create mode 100644 nvchecker_source/alpm.py
 create mode 100644 tests/test_alpm.py

diff --git a/docs/usage.rst b/docs/usage.rst
index 6e0a298..5ae2b39 100644
--- a/docs/usage.rst
+++ b/docs/usage.rst
@@ -689,6 +689,29 @@ registry
 
 This source returns tags and supports :ref:`list options`.
 
+Check ALPM database
+~~~~~~~~~~~~~~~~~~~
+::
+
+  source = "alpm"
+
+Check package updates in a local ALPM database.
+
+alpm
+  Name of the package.
+
+repo
+  Name of the package repository in which the package resides.
+
+dbpath
+  Path to the ALPM database directory. Default: ``/var/lib/pacman``.
+
+strip_release
+  Strip the release part, only return the part before ``-``.
+
+provided
+  Instead of the package version, return the version this package provides. Its value is what the package provides, and ``strip_release`` takes effect too. This is best used with libraries.
+
 Manually updating
 ~~~~~~~~~~~~~~~~~
 ::
diff --git a/nvchecker_source/alpm.py b/nvchecker_source/alpm.py
new file mode 100644
index 0000000..8c9a46f
--- /dev/null
+++ b/nvchecker_source/alpm.py
@@ -0,0 +1,34 @@
+# MIT licensed
+# Copyright (c) 2020 DDoSolitary <DDoSolitary@gmail.com>, et al.
+
+from nvchecker.api import GetVersionError
+from pyalpm import Handle
+
+
+async def open_db(info):
+  dbpath, repo = info
+  handle = Handle('/', dbpath)
+  db = handle.register_syncdb(repo, 0)
+  return (handle, db)
+
+
+async def get_version(name, conf, *, cache, **kwargs):
+  pkgname = conf.get('alpm', name)
+  dbpath = conf.get('dbpath', '/var/lib/pacman')
+  repo = conf.get('repo')
+  strip_release = conf.get('strip_release', False)
+  provided = conf.get('provided')
+  db = (await cache.get((dbpath, repo), open_db))[1]
+  pkg = db.get_pkg(pkgname)
+  if pkg is None:
+    raise GetVersionError('package not found in the ALPM database')
+  if provided is None:
+    version = pkg.version
+  else:
+    provides = dict(x.split('=', 1) for x in pkg.provides if '=' in x)
+    version = provides.get(provided)
+    if version is None:
+      raise GetVersionError('provides element not found')
+  if strip_release:
+    version = version.split('-', 1)[0]
+  return version
diff --git a/tests/test_alpm.py b/tests/test_alpm.py
new file mode 100644
index 0000000..ac64f22
--- /dev/null
+++ b/tests/test_alpm.py
@@ -0,0 +1,112 @@
+# MIT licensed
+# Copyright (c) 2020 DDoSolitary <DDoSolitary@gmail.com>, et al.
+
+import pathlib
+import pytest
+import os
+import shutil
+import subprocess
+import tempfile
+
+pytestmark = [
+  pytest.mark.asyncio,
+  pytest.mark.skipif(shutil.which('makepkg') is None, reason='requires makepkg command'),
+  pytest.mark.skipif(shutil.which('repo-add') is None, reason='requires repo-add command')
+]
+
+global temp_dir, db_path
+
+
+def setup_module(module):
+  global temp_dir, db_path
+  temp_dir = tempfile.TemporaryDirectory()
+  temp_path = pathlib.Path(temp_dir.name)
+  pkg_path = temp_path / 'test-pkg'
+  pkg_path.mkdir()
+  with (pkg_path / 'PKGBUILD').open('w') as f:
+    f.write(
+      'pkgname=test-pkg\n'
+      'pkgver=1.2.3\n'
+      'pkgrel=4\n'
+      'arch=(any)\n'
+      'provides=("test-provides=5.6-7" "test-provides-unversioned")\n'
+    )
+  subprocess.check_call(['makepkg'], cwd=pkg_path)
+  pkg_file = subprocess.check_output(['makepkg', '--packagelist'], cwd=pkg_path, text=True).strip()
+  db_path = pkg_path / 'test-db'
+  db_path.mkdir()
+  repo_path = db_path / 'sync'
+  repo_path.mkdir()
+  subprocess.check_call([
+    'repo-add',
+    repo_path / 'test-repo.db.tar.gz',
+    pkg_path / pkg_file
+  ])
+
+
+def teardown_module(module):
+  temp_dir.cleanup()
+
+
+async def test_alpm(get_version):
+  assert await get_version('test-pkg', {
+    'source': 'alpm',
+    'dbpath': str(db_path),
+    'repo': 'test-repo'
+  }) == '1.2.3-4'
+
+
+async def test_alpm_strip(get_version):
+  assert await get_version('test-pkg', {
+    'source': 'alpm',
+    'dbpath': str(db_path),
+    'repo': 'test-repo',
+    'strip_release': True
+  }) == '1.2.3'
+
+
+async def test_alpm_provided(get_version):
+  assert await get_version('test-pkg', {
+    'source': 'alpm',
+    'dbpath': str(db_path),
+    'repo': 'test-repo',
+    'provided': 'test-provides'
+  }) == '5.6-7'
+
+
+async def test_alpm_provided_strip(get_version):
+  assert await get_version('test-pkg', {
+    'source': 'alpm',
+    'dbpath': str(db_path),
+    'repo': 'test-repo',
+    'provided': 'test-provides',
+    'strip_release': True
+  }) == '5.6'
+
+
+async def test_alpm_missing_repo(get_version):
+  with pytest.raises(RuntimeError):
+    await get_version('test-pkg', {
+      'source': 'alpm',
+      'dbpath': str(db_path),
+      'repo': 'wrong-repo'
+    })
+
+
+async def test_alpm_missing_pkg(get_version):
+  with pytest.raises(RuntimeError):
+    await get_version('wrong-pkg', {
+      'source': 'alpm',
+      'dbpath': str(db_path),
+      'repo': 'test-repo'
+    })
+
+
+async def test_alpm_missing_provides(get_version):
+  with pytest.raises(RuntimeError):
+    await get_version('test-pkg', {
+      'source': 'alpm',
+      'dbpath': str(db_path),
+      'repo': 'test-repo',
+      'provided': 'wrong-provides'
+    })