[+] install command, main CLI wiring for network/cache/install

1. add cli/install.py: install_t class with --mode bash/exec,
     -m/--package-manager (pacman default), --dry-run BooleanOptionalAction,
     -r requirements, -d pkg_dir;
  2. add pacman_t.build_install_command() for generating pacman -U command;
  3. wire install, network, cache CLI into main.py;
  4. add test_install.py;

python/online/fxreader/pr34/commands_typed/archlinux/cli/install.py

@@ -0,0 +1,142 @@
+"""Install CLI: install downloaded packages via package manager."""
+
+import argparse
+import enum
+import logging
+import pathlib
+import shlex
+import subprocess
+
+from typing import Optional
+
+from ..apps.specs.models import compiled_entry_t
+from ..apps.specs.utils import parse_compiled
+
+logger = logging.getLogger(__name__)
+
+
+class install_t:
+	class mode_t(enum.StrEnum):
+		bash = 'bash'
+		exec = 'exec'
+
+	class package_manager_t(enum.StrEnum):
+		pacman = 'pacman'
+
+	def __init__(self) -> None:
+		self._options: Optional[argparse.Namespace] = None
+
+	@property
+	def options(self) -> argparse.Namespace:
+		assert self._options is not None
+		return self._options
+
+	def build_parser(self) -> argparse.ArgumentParser:
+		p = argparse.ArgumentParser(
+			prog='online-fxreader-pr34-archlinux install',
+			description='Install downloaded packages from a compiled requirements file.',
+		)
+		p.add_argument(
+			'-r',
+			dest='requirements',
+			required=True,
+			help='path to compiled requirements file',
+		)
+		p.add_argument(
+			'-d',
+			dest='pkg_dir',
+			required=True,
+			help='directory containing downloaded .pkg.tar.zst files',
+		)
+		p.add_argument(
+			'--mode',
+			choices=[m.value for m in install_t.mode_t],
+			default=install_t.mode_t.bash.value,
+			help='bash (default) = print the install command; '
+			'exec = execute the install command',
+		)
+		p.add_argument(
+			'-m', '--package-manager',
+			dest='package_manager',
+			choices=[m.value for m in install_t.package_manager_t],
+			default=install_t.package_manager_t.pacman.value,
+			help='package manager to use (default: pacman)',
+		)
+		p.add_argument(
+			'--dry-run',
+			action=argparse.BooleanOptionalAction,
+			default=True,
+			help='dry run: show what would happen without executing (default: true)',
+		)
+		return p
+
+	def parse_requirements(self, path: pathlib.Path) -> list[compiled_entry_t]:
+		return parse_compiled(path.read_text())
+
+	def resolve_paths(self, entries: list[compiled_entry_t], pkg_dir: pathlib.Path) -> list[pathlib.Path]:
+		paths: list[pathlib.Path] = []
+		for e in entries:
+			if e.filename == '':
+				raise FileNotFoundError(
+					'no filename for package %s==%s' % (e.name, e.version)
+				)
+			p = pkg_dir / e.filename
+			if not p.exists():
+				raise FileNotFoundError(
+					'package file not found: %s (expected at %s)' % (e.filename, p)
+				)
+			paths.append(p)
+		return paths
+
+	def build_command(self, paths: list[pathlib.Path]) -> list[str]:
+		pm = install_t.package_manager_t(self.options.package_manager)
+		if pm is install_t.package_manager_t.pacman:
+			from ..apps.pacman.client import pacman_t
+
+			return pacman_t.build_install_command(paths)
+		raise NotImplementedError('package manager: %s' % pm)
+
+	def run(self, args: list[str]) -> int:
+		parser = self.build_parser()
+		self._options = parser.parse_args(args)
+
+		req_path = pathlib.Path(self.options.requirements)
+		pkg_dir = pathlib.Path(self.options.pkg_dir)
+
+		entries = self.parse_requirements(req_path)
+		if len(entries) == 0:
+			logger.warning(dict(msg='no packages found in requirements file'))
+			return 0
+
+		try:
+			paths = self.resolve_paths(entries, pkg_dir)
+		except FileNotFoundError as e:
+			logger.error(str(e))
+			return 1
+
+		cmd = self.build_command(paths)
+		mode = install_t.mode_t(self.options.mode)
+
+		if self.options.dry_run:
+			logger.info(dict(
+				msg='dry run',
+				mode=mode.value,
+				packages=len(paths),
+				command=' '.join(shlex.quote(c) for c in cmd),
+			))
+			return 0
+
+		if mode is install_t.mode_t.bash:
+			print(' '.join(shlex.quote(c) for c in cmd))
+			return 0
+
+		if mode is install_t.mode_t.exec:
+			logger.info(dict(msg='executing', packages=len(paths)))
+			subprocess.check_call(cmd)
+			return 0
+
+		raise NotImplementedError('mode: %s' % mode)
+
+
+def main(args: list[str]) -> int:
+	return install_t().run(args)

python/online/fxreader/pr34/commands_typed/archlinux/cli/main.py

@@ -18,6 +18,7 @@ class Command(enum.Enum):
 	list_installed = 'list-installed'
 	compile = 'compile'
 	download = 'download'
+	install = 'install'
 	archive = 'archive'
 	diff = 'diff'
 	cve = 'cve'
@@ -44,6 +45,12 @@ def main(argv: Optional[list[str]] = None) -> int:
 		help='log level (default: INFO)',
 	)
 
+	from ..apps.network.cli import net_cli_t
+	from ..apps.cache.cli import cache_cli_t
+
+	net_cli_t.add_arguments(parser)
+	cache_cli_t.add_arguments(parser)
+
 	options, args = pr34_argparse.parse_args(parser, argv, stop_at=command_values)
 
 	if len(args) == 0 or args[0] not in command_values:
@@ -53,6 +60,9 @@ def main(argv: Optional[list[str]] = None) -> int:
 	options.command = Command(args[0])
 	args = args[1:]
 
+	net_cli_t.apply(net_cli_t.extract(options))
+	cache_cli_t.apply(cache_cli_t.extract(options))
+
 	pr34_logging.setup(
 		level=getattr(logging, options.log_level),
 		format=pr34_logging.format_t.json,
@@ -71,6 +81,10 @@ def main(argv: Optional[list[str]] = None) -> int:
 		from . import download
 
 		return download.main(args)
+	elif options.command is Command.install:
+		from . import install
+
+		return install.main(args)
 	elif options.command is Command.archive:
 		from . import archive
 

python/online/fxreader/pr34/commands_typed/archlinux/tests/test_install.py

@@ -0,0 +1,138 @@
+import pathlib
+import tempfile
+import unittest
+import unittest.mock
+
+from ..cli.install import install_t
+from ..apps.pacman.client import pacman_t
+
+
+class TestInstall(unittest.TestCase):
+	def setUp(self) -> None:
+		self._tmp = tempfile.TemporaryDirectory()
+		self.tmp = pathlib.Path(self._tmp.name)
+
+	def tearDown(self) -> None:
+		self._tmp.cleanup()
+
+	def _write_requirements(self, lines: list[str]) -> pathlib.Path:
+		p = self.tmp / 'requirements.txt'
+		p.write_text('\n'.join(lines) + '\n')
+		return p
+
+	def _touch_pkg(self, filename: str) -> pathlib.Path:
+		pkg_dir = self.tmp / 'pkgs'
+		pkg_dir.mkdir(exist_ok=True)
+		p = pkg_dir / filename
+		p.write_bytes(b'fake')
+		return p
+
+	def test_parse_requirements(self) -> None:
+		req = self._write_requirements([
+			'# https://archive.archlinux.org/packages/c/crun/crun-1.27-1-x86_64.pkg.tar.zst',
+			'crun==1.27-1 --hash=sha256:abc --size=12345',
+			'# https://archive.archlinux.org/packages/y/yajl/yajl-2.1.0-6-x86_64.pkg.tar.zst',
+			'yajl==2.1.0-6 --hash=sha256:def --size=9999',
+		])
+		inst = install_t()
+		entries = inst.parse_requirements(req)
+		self.assertEqual(len(entries), 2)
+		self.assertEqual(entries[0].name, 'crun')
+		self.assertEqual(entries[0].filename, 'crun-1.27-1-x86_64.pkg.tar.zst')
+		self.assertEqual(entries[1].name, 'yajl')
+
+	def test_resolve_paths(self) -> None:
+		self._touch_pkg('crun-1.27-1-x86_64.pkg.tar.zst')
+		self._touch_pkg('yajl-2.1.0-6-x86_64.pkg.tar.zst')
+
+		req = self._write_requirements([
+			'# https://archive.archlinux.org/packages/c/crun/crun-1.27-1-x86_64.pkg.tar.zst',
+			'crun==1.27-1 --hash=sha256:abc --size=12345',
+			'# https://archive.archlinux.org/packages/y/yajl/yajl-2.1.0-6-x86_64.pkg.tar.zst',
+			'yajl==2.1.0-6 --hash=sha256:def --size=9999',
+		])
+		inst = install_t()
+		entries = inst.parse_requirements(req)
+		pkg_dir = self.tmp / 'pkgs'
+		paths = inst.resolve_paths(entries, pkg_dir)
+		self.assertEqual(len(paths), 2)
+		for p in paths:
+			self.assertTrue(p.exists())
+
+	def test_resolve_paths_missing_raises(self) -> None:
+		req = self._write_requirements([
+			'# https://archive.archlinux.org/packages/c/crun/crun-1.27-1-x86_64.pkg.tar.zst',
+			'crun==1.27-1 --hash=sha256:abc --size=12345',
+		])
+		inst = install_t()
+		entries = inst.parse_requirements(req)
+		pkg_dir = self.tmp / 'pkgs'
+		pkg_dir.mkdir(exist_ok=True)
+		with self.assertRaises(FileNotFoundError):
+			inst.resolve_paths(entries, pkg_dir)
+
+	def test_build_command(self) -> None:
+		paths = [
+			pathlib.Path('/tmp/pkgs/crun-1.27-1-x86_64.pkg.tar.zst'),
+			pathlib.Path('/tmp/pkgs/yajl-2.1.0-6-x86_64.pkg.tar.zst'),
+		]
+		cmd = pacman_t.build_install_command(paths)
+		self.assertEqual(cmd[0], 'pacman')
+		self.assertIn('-U', cmd)
+		self.assertIn('--noconfirm', cmd)
+		for p in paths:
+			self.assertIn(str(p), cmd)
+
+	def test_dry_run_does_not_execute(self) -> None:
+		self._touch_pkg('crun-1.27-1-x86_64.pkg.tar.zst')
+		req = self._write_requirements([
+			'# https://archive.archlinux.org/packages/c/crun/crun-1.27-1-x86_64.pkg.tar.zst',
+			'crun==1.27-1 --hash=sha256:abc --size=12345',
+		])
+		pkg_dir = self.tmp / 'pkgs'
+		with unittest.mock.patch('subprocess.check_call') as mock_call:
+			ret = install_t().run([
+				'-r', str(req),
+				'-d', str(pkg_dir),
+				'--dry-run',
+			])
+		mock_call.assert_not_called()
+		self.assertEqual(ret, 0)
+
+	def test_mode_bash_outputs_command(self) -> None:
+		self._touch_pkg('crun-1.27-1-x86_64.pkg.tar.zst')
+		req = self._write_requirements([
+			'# https://archive.archlinux.org/packages/c/crun/crun-1.27-1-x86_64.pkg.tar.zst',
+			'crun==1.27-1 --hash=sha256:abc --size=12345',
+		])
+		pkg_dir = self.tmp / 'pkgs'
+		with unittest.mock.patch('builtins.print') as mock_print:
+			ret = install_t().run([
+				'-r', str(req),
+				'-d', str(pkg_dir),
+				'--mode', 'bash',
+				'--no-dry-run',
+			])
+		self.assertEqual(ret, 0)
+		output = mock_print.call_args[0][0]
+		self.assertIn('pacman', output)
+		self.assertIn('-U', output)
+
+	def test_mode_exec_calls_subprocess(self) -> None:
+		self._touch_pkg('crun-1.27-1-x86_64.pkg.tar.zst')
+		req = self._write_requirements([
+			'# https://archive.archlinux.org/packages/c/crun/crun-1.27-1-x86_64.pkg.tar.zst',
+			'crun==1.27-1 --hash=sha256:abc --size=12345',
+		])
+		pkg_dir = self.tmp / 'pkgs'
+		with unittest.mock.patch('subprocess.check_call') as mock_call:
+			ret = install_t().run([
+				'-r', str(req),
+				'-d', str(pkg_dir),
+				'--mode', 'exec',
+				'--no-dry-run',
+			])
+		self.assertEqual(ret, 0)
+		mock_call.assert_called_once()
+		cmd = mock_call.call_args[0][0]
+		self.assertEqual(cmd[0], 'pacman')