diff --git a/python/online/fxreader/pr34/commands_typed/archlinux/cli/install.py b/python/online/fxreader/pr34/commands_typed/archlinux/cli/install.py new file mode 100644 index 0000000..ca0d35a --- /dev/null +++ b/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) diff --git a/python/online/fxreader/pr34/commands_typed/archlinux/cli/main.py b/python/online/fxreader/pr34/commands_typed/archlinux/cli/main.py index 9be8c31..c6de560 100644 --- a/python/online/fxreader/pr34/commands_typed/archlinux/cli/main.py +++ b/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 diff --git a/python/online/fxreader/pr34/commands_typed/archlinux/tests/test_install.py b/python/online/fxreader/pr34/commands_typed/archlinux/tests/test_install.py new file mode 100644 index 0000000..1b9f320 --- /dev/null +++ b/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')