diff --git a/Makefile b/Makefile index d482ad6..5849669 100644 --- a/Makefile +++ b/Makefile @@ -68,8 +68,8 @@ test-container: build-missing build-missing: @bash scripts/build/build-image-missing.sh -# Combined test target for local + CI (unit + e2e + integration) -test: test-container test-unit test-e2e test-integration +# Combined test target for local + CI (unit + integration + e2e) +test: test-container test-unit test-integration test-e2e # ------------------------------------------------------------ # System install (native packages, calls scripts/installation/run-package.sh) diff --git a/tests/unit/pkgmgr/actions/install/installers/test_nix_flake.py b/tests/unit/pkgmgr/actions/install/installers/test_nix_flake.py index 96329c9..b49f346 100644 --- a/tests/unit/pkgmgr/actions/install/installers/test_nix_flake.py +++ b/tests/unit/pkgmgr/actions/install/installers/test_nix_flake.py @@ -1,18 +1,22 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + import os import unittest from unittest import mock -from unittest.mock import patch +from unittest.mock import MagicMock, patch from pkgmgr.actions.repository.install.context import RepoContext from pkgmgr.actions.repository.install.installers.nix_flake import NixFlakeInstaller class TestNixFlakeInstaller(unittest.TestCase): - def setUp(self): - self.repo = {"name": "test-repo"} + def setUp(self) -> None: + self.repo = {"repository": "package-manager"} + # Important: identifier "pkgmgr" triggers both "pkgmgr" and "default" self.ctx = RepoContext( repo=self.repo, - identifier="test-id", + identifier="pkgmgr", repo_dir="/tmp/repo", repositories_base_dir="/tmp", bin_dir="/bin", @@ -25,99 +29,104 @@ class TestNixFlakeInstaller(unittest.TestCase): ) self.installer = NixFlakeInstaller() - @patch("shutil.which", return_value="/usr/bin/nix") - @patch("os.path.exists", return_value=True) - def test_supports_true_when_nix_and_flake_exist(self, mock_exists, mock_which): - """ - supports() should return True when: - - nix is available, - - flake.nix exists in the repo, - - and we are not inside a Nix dev shell. - """ - with patch.dict(os.environ, {"IN_NIX_SHELL": ""}, clear=False): + @patch("pkgmgr.actions.repository.install.installers.nix_flake.os.path.exists") + @patch("pkgmgr.actions.repository.install.installers.nix_flake.shutil.which") + def test_supports_true_when_nix_and_flake_exist( + self, + mock_which: MagicMock, + mock_exists: MagicMock, + ) -> None: + mock_which.return_value = "/usr/bin/nix" + mock_exists.return_value = True + + with patch.dict(os.environ, {"PKGMGR_DISABLE_NIX_FLAKE_INSTALLER": ""}, clear=False): self.assertTrue(self.installer.supports(self.ctx)) - mock_which.assert_called_with("nix") - mock_exists.assert_called_with(os.path.join(self.ctx.repo_dir, "flake.nix")) + mock_which.assert_called_once_with("nix") + mock_exists.assert_called_once_with( + os.path.join(self.ctx.repo_dir, self.installer.FLAKE_FILE) + ) - @patch("shutil.which", return_value=None) - @patch("os.path.exists", return_value=True) - def test_supports_false_when_nix_missing(self, mock_exists, mock_which): - """ - supports() should return False if nix is not available, - even if a flake.nix file exists. - """ - with patch.dict(os.environ, {"IN_NIX_SHELL": ""}, clear=False): + @patch("pkgmgr.actions.repository.install.installers.nix_flake.os.path.exists") + @patch("pkgmgr.actions.repository.install.installers.nix_flake.shutil.which") + def test_supports_false_when_nix_missing( + self, + mock_which: MagicMock, + mock_exists: MagicMock, + ) -> None: + mock_which.return_value = None + mock_exists.return_value = True # flake exists but nix is missing + + with patch.dict(os.environ, {"PKGMGR_DISABLE_NIX_FLAKE_INSTALLER": ""}, clear=False): self.assertFalse(self.installer.supports(self.ctx)) - @patch("os.path.exists", return_value=True) - @patch("shutil.which", return_value="/usr/bin/nix") - @mock.patch("pkgmgr.actions.repository.install.installers.nix_flake.run_command") + @patch("pkgmgr.actions.repository.install.installers.nix_flake.os.path.exists") + @patch("pkgmgr.actions.repository.install.installers.nix_flake.shutil.which") + def test_supports_false_when_disabled_via_env( + self, + mock_which: MagicMock, + mock_exists: MagicMock, + ) -> None: + mock_which.return_value = "/usr/bin/nix" + mock_exists.return_value = True + + with patch.dict( + os.environ, + {"PKGMGR_DISABLE_NIX_FLAKE_INSTALLER": "1"}, + clear=False, + ): + self.assertFalse(self.installer.supports(self.ctx)) + + @patch("pkgmgr.actions.repository.install.installers.nix_flake.NixFlakeInstaller.supports") + @patch("pkgmgr.actions.repository.install.installers.nix_flake.run_command") def test_run_removes_old_profile_and_installs_outputs( self, - mock_run_command, - mock_which, - mock_exists, - ): + mock_run_command: MagicMock, + mock_supports: MagicMock, + ) -> None: """ run() should: - - 1. attempt to remove the old 'package-manager' profile entry, and - 2. install both 'pkgmgr' and 'default' flake outputs. + - remove the old profile + - install both 'pkgmgr' and 'default' outputs for identifier 'pkgmgr' + - call commands in the correct order """ + mock_supports.return_value = True - cmds = [] + commands: list[str] = [] - def side_effect(cmd, cwd=None, preview=False, *args, **kwargs): - cmds.append(cmd) - return None + def side_effect(cmd: str, cwd: str | None = None, preview: bool = False, **_: object) -> None: + commands.append(cmd) mock_run_command.side_effect = side_effect - # Simulate a normal environment (not inside nix develop, installer enabled). - with patch.dict( - os.environ, - {"IN_NIX_SHELL": "", "PKGMGR_DISABLE_NIX_FLAKE_INSTALLER": ""}, - clear=False, - ): + with patch.dict(os.environ, {"PKGMGR_DISABLE_NIX_FLAKE_INSTALLER": ""}, clear=False): self.installer.run(self.ctx) remove_cmd = f"nix profile remove {self.installer.PROFILE_NAME} || true" install_pkgmgr_cmd = f"nix profile install {self.ctx.repo_dir}#pkgmgr" install_default_cmd = f"nix profile install {self.ctx.repo_dir}#default" - # At least these three commands must have been issued. - self.assertIn(remove_cmd, cmds) - self.assertIn(install_pkgmgr_cmd, cmds) - self.assertIn(install_default_cmd, cmds) + self.assertIn(remove_cmd, commands) + self.assertIn(install_pkgmgr_cmd, commands) + self.assertIn(install_default_cmd, commands) - # Optional: ensure the remove call came first. - self.assertEqual(cmds[0], remove_cmd) + self.assertEqual(commands[0], remove_cmd) - @patch("shutil.which", return_value="/usr/bin/nix") - @mock.patch("pkgmgr.actions.repository.install.installers.nix_flake.run_command") + @patch("pkgmgr.actions.repository.install.installers.nix_flake.shutil.which") + @patch("pkgmgr.actions.repository.install.installers.nix_flake.run_command") def test_ensure_old_profile_removed_ignores_systemexit( self, - mock_run_command, - mock_which, - ): - """ - _ensure_old_profile_removed() must not propagate SystemExit, even if - 'nix profile remove' fails (e.g. profile entry does not exist). - """ + mock_run_command: MagicMock, + mock_which: MagicMock, + ) -> None: + mock_which.return_value = "/usr/bin/nix" - def side_effect(cmd, cwd=None, preview=False, *args, **kwargs): + def side_effect(cmd: str, cwd: str | None = None, preview: bool = False, **_: object) -> None: raise SystemExit(1) mock_run_command.side_effect = side_effect - with patch.dict( - os.environ, - {"IN_NIX_SHELL": "", "PKGMGR_DISABLE_NIX_FLAKE_INSTALLER": ""}, - clear=False, - ): - # Should not raise, SystemExit is swallowed internally. - self.installer._ensure_old_profile_removed(self.ctx) + self.installer._ensure_old_profile_removed(self.ctx) remove_cmd = f"nix profile remove {self.installer.PROFILE_NAME} || true" mock_run_command.assert_called_with( diff --git a/tests/unit/pkgmgr/actions/install/test_install_repos.py b/tests/unit/pkgmgr/actions/install/test_install_repos.py index 8271c44..6c08cf3 100644 --- a/tests/unit/pkgmgr/actions/install/test_install_repos.py +++ b/tests/unit/pkgmgr/actions/install/test_install_repos.py @@ -1,134 +1,129 @@ -# tests/unit/pkgmgr/test_install_repos.py +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +import os import unittest -from unittest.mock import patch, MagicMock +from typing import Any, Dict, List +from unittest.mock import MagicMock, patch -from pkgmgr.actions.repository.install.context import RepoContext -import pkgmgr.actions.repository.install as install_module -from pkgmgr.actions.repository.install.installers.base import BaseInstaller +from pkgmgr.actions.repository.install import install_repos -class DummyInstaller(BaseInstaller): - """Simple installer for testing orchestration.""" - - layer = None # no specific capabilities - - def __init__(self): - self.calls = [] - - def supports(self, ctx: RepoContext) -> bool: - # Always support to verify that the pipeline runs - return True - - def run(self, ctx: RepoContext) -> None: - self.calls.append(ctx.identifier) +Repository = Dict[str, Any] class TestInstallReposOrchestration(unittest.TestCase): - @patch("pkgmgr.actions.repository.install.create_ink") - @patch("pkgmgr.actions.repository.install.resolve_command_for_repo") - @patch("pkgmgr.actions.repository.install.verify_repository") - @patch("pkgmgr.actions.repository.install.get_repo_dir") - @patch("pkgmgr.actions.repository.install.get_repo_identifier") + def setUp(self) -> None: + self.base_dir = "/fake/base" + self.bin_dir = "/fake/bin" + + self.repo1: Repository = { + "account": "kevinveenbirkenbach", + "repository": "repo-one", + "alias": "repo-one", + "verified": {"gpg_keys": ["FAKEKEY"]}, + } + self.repo2: Repository = { + "account": "kevinveenbirkenbach", + "repository": "repo-two", + "alias": "repo-two", + "verified": {"gpg_keys": ["FAKEKEY"]}, + } + self.all_repos: List[Repository] = [self.repo1, self.repo2] + + @patch("pkgmgr.actions.repository.install.InstallationPipeline") @patch("pkgmgr.actions.repository.install.clone_repos") + @patch("pkgmgr.actions.repository.install.get_repo_dir") + @patch("pkgmgr.actions.repository.install.os.path.exists", return_value=True) + @patch( + "pkgmgr.actions.repository.install.verify_repository", + return_value=(True, [], "hash", "key"), + ) def test_install_repos_runs_pipeline_for_each_repo( self, - mock_clone_repos, - mock_get_repo_identifier, - mock_get_repo_dir, - mock_verify_repository, - mock_resolve_command_for_repo, - mock_create_ink, - ): - repo1 = {"name": "repo1"} - repo2 = {"name": "repo2"} - selected_repos = [repo1, repo2] - all_repos = selected_repos + _mock_verify_repository: MagicMock, + _mock_exists: MagicMock, + mock_get_repo_dir: MagicMock, + mock_clone_repos: MagicMock, + mock_pipeline_cls: MagicMock, + ) -> None: + """ + install_repos() should construct a RepoContext for each repository and + run the InstallationPipeline exactly once per selected repo when the + repo directory exists and verification passes. + """ + mock_get_repo_dir.side_effect = [ + os.path.join(self.base_dir, "repo-one"), + os.path.join(self.base_dir, "repo-two"), + ] - # Return identifiers and directories - mock_get_repo_identifier.side_effect = ["id1", "id2"] - mock_get_repo_dir.side_effect = ["/tmp/repo1", "/tmp/repo2"] + selected = [self.repo1, self.repo2] - # Simulate verification success: (ok, errors, commit, key) - mock_verify_repository.return_value = (True, [], "commit", "key") + install_repos( + selected_repos=selected, + repositories_base_dir=self.base_dir, + bin_dir=self.bin_dir, + all_repos=self.all_repos, + no_verification=False, + preview=False, + quiet=False, + clone_mode="ssh", + update_dependencies=False, + ) - # Resolve commands for both repos so create_ink will be called - mock_resolve_command_for_repo.side_effect = ["/bin/cmd1", "/bin/cmd2"] + # clone_repos must not be called because directories "exist" + mock_clone_repos.assert_not_called() - # Ensure directories exist (no cloning) - with patch("os.path.exists", return_value=True): - dummy_installer = DummyInstaller() - # Monkeypatch INSTALLERS for this test - old_installers = install_module.INSTALLERS - install_module.INSTALLERS = [dummy_installer] - try: - install_module.install_repos( - selected_repos=selected_repos, - repositories_base_dir="/tmp", - bin_dir="/bin", - all_repos=all_repos, - no_verification=False, - preview=False, - quiet=False, - clone_mode="ssh", - update_dependencies=False, - ) - finally: - install_module.INSTALLERS = old_installers + # A pipeline is constructed once, then run() is invoked once per repo + self.assertEqual(mock_pipeline_cls.call_count, 1) + pipeline_instance = mock_pipeline_cls.return_value + self.assertEqual(pipeline_instance.run.call_count, len(selected)) - # Check that installers ran with both identifiers - self.assertEqual(dummy_installer.calls, ["id1", "id2"]) - self.assertEqual(mock_create_ink.call_count, 2) - self.assertEqual(mock_verify_repository.call_count, 2) - self.assertEqual(mock_resolve_command_for_repo.call_count, 2) - - @patch("pkgmgr.actions.repository.install.verify_repository") - @patch("pkgmgr.actions.repository.install.get_repo_dir") - @patch("pkgmgr.actions.repository.install.get_repo_identifier") + @patch("pkgmgr.actions.repository.install.InstallationPipeline") @patch("pkgmgr.actions.repository.install.clone_repos") + @patch("pkgmgr.actions.repository.install.get_repo_dir") + @patch("pkgmgr.actions.repository.install.os.path.exists", return_value=True) + @patch( + "pkgmgr.actions.repository.install.verify_repository", + return_value=(False, ["invalid signature"], None, None), + ) + @patch("builtins.input", return_value="n") def test_install_repos_skips_on_failed_verification( self, - mock_clone_repos, - mock_get_repo_identifier, - mock_get_repo_dir, - mock_verify_repository, - ): - repo = {"name": "repo1", "verified": True} - selected_repos = [repo] - all_repos = selected_repos + _mock_input: MagicMock, + _mock_verify_repository: MagicMock, + _mock_exists: MagicMock, + mock_get_repo_dir: MagicMock, + mock_clone_repos: MagicMock, + mock_pipeline_cls: MagicMock, + ) -> None: + """ + When verification fails and the user does NOT confirm installation, + the InstallationPipeline must not be run for that repository. + """ + mock_get_repo_dir.return_value = os.path.join(self.base_dir, "repo-one") - mock_get_repo_identifier.return_value = "id1" - mock_get_repo_dir.return_value = "/tmp/repo1" + selected = [self.repo1] - # Verification fails: ok=False, with error list - mock_verify_repository.return_value = (False, ["sig error"], None, None) + install_repos( + selected_repos=selected, + repositories_base_dir=self.base_dir, + bin_dir=self.bin_dir, + all_repos=self.all_repos, + no_verification=False, + preview=False, + quiet=False, + clone_mode="ssh", + update_dependencies=False, + ) - dummy_installer = DummyInstaller() - with patch("pkgmgr.actions.repository.install.create_ink") as mock_create_ink, \ - patch("pkgmgr.actions.repository.install.resolve_command_for_repo") as mock_resolve_cmd, \ - patch("os.path.exists", return_value=True), \ - patch("builtins.input", return_value="n"): - old_installers = install_module.INSTALLERS - install_module.INSTALLERS = [dummy_installer] - try: - install_module.install_repos( - selected_repos=selected_repos, - repositories_base_dir="/tmp", - bin_dir="/bin", - all_repos=all_repos, - no_verification=False, - preview=False, - quiet=False, - clone_mode="ssh", - update_dependencies=False, - ) - finally: - install_module.INSTALLERS = old_installers + # clone_repos must not be called because directory "exists" + mock_clone_repos.assert_not_called() - # No installer run and no create_ink when user declines - self.assertEqual(dummy_installer.calls, []) - mock_create_ink.assert_not_called() - mock_resolve_cmd.assert_not_called() + # Pipeline is constructed, but run() must not be called + mock_pipeline_cls.assert_called_once() + pipeline_instance = mock_pipeline_cls.return_value + pipeline_instance.run.assert_not_called() if __name__ == "__main__": diff --git a/tests/unit/pkgmgr/actions/install/test_layers.py b/tests/unit/pkgmgr/actions/install/test_layers.py new file mode 100644 index 0000000..1bbafae --- /dev/null +++ b/tests/unit/pkgmgr/actions/install/test_layers.py @@ -0,0 +1,94 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +import os +import unittest + +from pkgmgr.actions.repository.install.layers import ( + CliLayer, + CLI_LAYERS, + classify_command_layer, + layer_priority, +) + + +class TestCliLayerAndPriority(unittest.TestCase): + def test_layer_priority_for_known_layers_is_monotonic(self) -> None: + """ + layer_priority() must reflect the ordering in CLI_LAYERS. + We mainly check that the order is stable and that each later item + has a higher (or equal) priority index than the previous one. + """ + priorities = [layer_priority(layer) for layer in CLI_LAYERS] + + # Ensure no negative priorities and strictly increasing or stable order + for idx, value in enumerate(priorities): + self.assertGreaterEqual( + value, 0, f"Priority for {CLI_LAYERS[idx]} must be >= 0" + ) + if idx > 0: + self.assertGreaterEqual( + value, + priorities[idx - 1], + "Priorities must be non-decreasing in CLI_LAYERS order", + ) + + def test_layer_priority_for_none_and_unknown(self) -> None: + """ + None and unknown layers should both receive the 'max' priority + (i.e., len(CLI_LAYERS)). + """ + none_priority = layer_priority(None) + self.assertEqual(none_priority, len(CLI_LAYERS)) + + class FakeLayer: + # Not part of CliLayer + pass + + unknown_priority = layer_priority(FakeLayer()) # type: ignore[arg-type] + self.assertEqual(unknown_priority, len(CLI_LAYERS)) + + +class TestClassifyCommandLayer(unittest.TestCase): + def setUp(self) -> None: + self.home = os.path.expanduser("~") + self.repo_dir = "/tmp/pkgmgr-test-repo" + + def test_classify_system_binaries_os_packages(self) -> None: + for cmd in ("/usr/bin/pkgmgr", "/bin/pkgmgr"): + with self.subTest(cmd=cmd): + layer = classify_command_layer(cmd, self.repo_dir) + self.assertEqual(layer, CliLayer.OS_PACKAGES) + + def test_classify_nix_binaries(self) -> None: + nix_cmds = [ + "/nix/store/abcd1234-bin-pkgmgr/bin/pkgmgr", + os.path.join(self.home, ".nix-profile", "bin", "pkgmgr"), + ] + for cmd in nix_cmds: + with self.subTest(cmd=cmd): + layer = classify_command_layer(cmd, self.repo_dir) + self.assertEqual(layer, CliLayer.NIX) + + def test_classify_python_binaries(self) -> None: + # Default Python/virtualenv-style location in home + cmd = os.path.join(self.home, ".local", "bin", "pkgmgr") + layer = classify_command_layer(cmd, self.repo_dir) + self.assertEqual(layer, CliLayer.PYTHON) + + def test_classify_repo_local_binary_makefile_layer(self) -> None: + cmd = os.path.join(self.repo_dir, "bin", "pkgmgr") + layer = classify_command_layer(cmd, self.repo_dir) + self.assertEqual(layer, CliLayer.MAKEFILE) + + def test_fallback_to_python_layer(self) -> None: + """ + Non-system, non-nix, non-repo binaries should fall back to PYTHON. + """ + cmd = "/opt/pkgmgr/bin/pkgmgr" + layer = classify_command_layer(cmd, self.repo_dir) + self.assertEqual(layer, CliLayer.PYTHON) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/pkgmgr/actions/install/test_pipeline.py b/tests/unit/pkgmgr/actions/install/test_pipeline.py new file mode 100644 index 0000000..84195eb --- /dev/null +++ b/tests/unit/pkgmgr/actions/install/test_pipeline.py @@ -0,0 +1,157 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +import unittest +from unittest.mock import MagicMock, patch + +from pkgmgr.actions.repository.install.context import RepoContext +from pkgmgr.actions.repository.install.installers.base import BaseInstaller +from pkgmgr.actions.repository.install.layers import CliLayer +from pkgmgr.actions.repository.install.pipeline import InstallationPipeline + + +class DummyInstaller(BaseInstaller): + """ + Small fake installer with configurable layer, supports() result, + and advertised capabilities. + """ + + def __init__( + self, + name: str, + layer: str | None = None, + supports_result: bool = True, + capabilities: set[str] | None = None, + ) -> None: + self._name = name + self.layer = layer # type: ignore[assignment] + self._supports_result = supports_result + self._capabilities = capabilities or set() + self.ran = False + + def supports(self, ctx: RepoContext) -> bool: # type: ignore[override] + return self._supports_result + + def run(self, ctx: RepoContext) -> None: # type: ignore[override] + self.ran = True + + def discover_capabilities(self, ctx: RepoContext) -> set[str]: # type: ignore[override] + return set(self._capabilities) + + +def _minimal_context() -> RepoContext: + repo = { + "account": "kevinveenbirkenbach", + "repository": "test-repo", + "alias": "test-repo", + } + return RepoContext( + repo=repo, + identifier="test-repo", + repo_dir="/tmp/test-repo", + repositories_base_dir="/tmp", + bin_dir="/usr/local/bin", + all_repos=[repo], + no_verification=False, + preview=False, + quiet=False, + clone_mode="ssh", + update_dependencies=False, + ) + + +class TestInstallationPipeline(unittest.TestCase): + @patch("pkgmgr.actions.repository.install.pipeline.create_ink") + @patch("pkgmgr.actions.repository.install.pipeline.resolve_command_for_repo") + def test_create_ink_called_when_command_resolved( + self, + mock_resolve_command_for_repo: MagicMock, + mock_create_ink: MagicMock, + ) -> None: + """ + If resolve_command_for_repo returns a command, InstallationPipeline + must attach it to the repo and call create_ink(). + """ + mock_resolve_command_for_repo.return_value = "/usr/local/bin/test-repo" + + ctx = _minimal_context() + installer = DummyInstaller("noop-installer", supports_result=False) + pipeline = InstallationPipeline([installer]) + + pipeline.run(ctx) + + self.assertTrue(mock_create_ink.called) + self.assertEqual( + ctx.repo.get("command"), + "/usr/local/bin/test-repo", + ) + + @patch("pkgmgr.actions.repository.install.pipeline.create_ink") + @patch("pkgmgr.actions.repository.install.pipeline.resolve_command_for_repo") + def test_lower_priority_installers_are_skipped_if_cli_exists( + self, + mock_resolve_command_for_repo: MagicMock, + mock_create_ink: MagicMock, + ) -> None: + """ + If the resolved command is provided by a higher-priority layer + (e.g. OS_PACKAGES), a lower-priority installer (e.g. PYTHON) + must be skipped. + """ + mock_resolve_command_for_repo.return_value = "/usr/bin/test-repo" + + ctx = _minimal_context() + python_installer = DummyInstaller( + "python-installer", + layer=CliLayer.PYTHON.value, + supports_result=True, + ) + pipeline = InstallationPipeline([python_installer]) + + pipeline.run(ctx) + + self.assertFalse( + python_installer.ran, + "Python installer must not run when an OS_PACKAGES CLI already exists.", + ) + self.assertEqual(ctx.repo.get("command"), "/usr/bin/test-repo") + + @patch("pkgmgr.actions.repository.install.pipeline.create_ink") + @patch("pkgmgr.actions.repository.install.pipeline.resolve_command_for_repo") + def test_capabilities_prevent_duplicate_installers( + self, + mock_resolve_command_for_repo: MagicMock, + mock_create_ink: MagicMock, + ) -> None: + """ + If one installer has already provided a set of capabilities, + a second installer advertising the same capabilities should be skipped. + """ + mock_resolve_command_for_repo.return_value = None # no CLI initially + + ctx = _minimal_context() + first = DummyInstaller( + "first-installer", + layer=CliLayer.PYTHON.value, + supports_result=True, + capabilities={"cli"}, + ) + second = DummyInstaller( + "second-installer", + layer=CliLayer.PYTHON.value, + supports_result=True, + capabilities={"cli"}, # same capability + ) + + pipeline = InstallationPipeline([first, second]) + pipeline.run(ctx) + + self.assertTrue(first.ran, "First installer should run.") + self.assertFalse( + second.ran, + "Second installer with identical capabilities must be skipped.", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/pkgmgr/core/command/test_resolve.py b/tests/unit/pkgmgr/core/command/test_resolve.py new file mode 100644 index 0000000..47cd539 --- /dev/null +++ b/tests/unit/pkgmgr/core/command/test_resolve.py @@ -0,0 +1,212 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +import os +import stat +import tempfile +import unittest +from unittest.mock import patch + +from pkgmgr.core.command.resolve import ( + _find_python_package_root, + _nix_binary_candidates, + _path_binary_candidates, + resolve_command_for_repo, +) + + +class TestHelpers(unittest.TestCase): + def test_find_python_package_root_none_when_missing_src(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = _find_python_package_root(tmpdir) + self.assertIsNone(root) + + def test_find_python_package_root_returns_existing_dir_or_none(self) -> None: + """ + We only assert that the helper does not return an invalid path. + The exact selection heuristic is intentionally left flexible since + the implementation may evolve. + """ + with tempfile.TemporaryDirectory() as tmpdir: + src_dir = os.path.join(tmpdir, "src", "mypkg") + os.makedirs(src_dir, exist_ok=True) + init_path = os.path.join(src_dir, "__init__.py") + with open(init_path, "w", encoding="utf-8") as f: + f.write("# package marker\n") + + root = _find_python_package_root(tmpdir) + if root is not None: + self.assertTrue(os.path.isdir(root)) + + def test_nix_binary_candidates_builds_expected_paths(self) -> None: + home = "/home/testuser" + names = ["pkgmgr", "", None, "other"] # type: ignore[list-item] + candidates = _nix_binary_candidates(home, names) # type: ignore[arg-type] + + self.assertIn( + os.path.join(home, ".nix-profile", "bin", "pkgmgr"), + candidates, + ) + self.assertIn( + os.path.join(home, ".nix-profile", "bin", "other"), + candidates, + ) + self.assertEqual(len(candidates), 2) + + @patch("pkgmgr.core.command.resolve._is_executable", return_value=True) + @patch("pkgmgr.core.command.resolve.shutil.which") + def test_path_binary_candidates_uses_which_and_executable( + self, + mock_which, + _mock_is_executable, + ) -> None: + def which_side_effect(name: str) -> str | None: + if name == "pkgmgr": + return "/usr/local/bin/pkgmgr" + if name == "other": + return "/usr/bin/other" + return None + + mock_which.side_effect = which_side_effect + + candidates = _path_binary_candidates(["pkgmgr", "other", "missing"]) + self.assertEqual( + candidates, + ["/usr/local/bin/pkgmgr", "/usr/bin/other"], + ) + + +class TestResolveCommandForRepo(unittest.TestCase): + def test_explicit_command_in_repo_wins(self) -> None: + repo = {"command": "/custom/path/pkgmgr"} + cmd = resolve_command_for_repo( + repo=repo, + repo_identifier="pkgmgr", + repo_dir="/tmp/pkgmgr", + ) + self.assertEqual(cmd, "/custom/path/pkgmgr") + + @patch("pkgmgr.core.command.resolve._is_executable", return_value=True) + @patch("pkgmgr.core.command.resolve._nix_binary_candidates", return_value=[]) + @patch("pkgmgr.core.command.resolve.shutil.which") + def test_prefers_non_system_path_over_system_binary( + self, + mock_which, + _mock_nix_candidates, + _mock_is_executable, + ) -> None: + """ + If both a system binary (/usr/bin) and a non-system binary (/opt/bin) + exist in PATH, the non-system binary must be preferred. + """ + + def which_side_effect(name: str) -> str | None: + if name == "pkgmgr": + return "/usr/bin/pkgmgr" # system binary + if name == "alias": + return "/opt/bin/pkgmgr" # non-system binary + return None + + mock_which.side_effect = which_side_effect + + repo = { + "alias": "alias", + "repository": "pkgmgr", + } + cmd = resolve_command_for_repo( + repo=repo, + repo_identifier="pkgmgr", + repo_dir="/tmp/pkgmgr", + ) + self.assertEqual(cmd, "/opt/bin/pkgmgr") + + @patch("pkgmgr.core.command.resolve._is_executable", return_value=True) + @patch("pkgmgr.core.command.resolve._nix_binary_candidates") + @patch("pkgmgr.core.command.resolve.shutil.which") + def test_nix_binary_used_when_no_non_system_bin( + self, + mock_which, + mock_nix_candidates, + _mock_is_executable, + ) -> None: + """ + When only a system binary exists in PATH but a Nix profile binary is + available, the Nix binary should be preferred. + """ + + def which_side_effect(name: str) -> str | None: + if name == "pkgmgr": + return "/usr/bin/pkgmgr" + return None + + mock_which.side_effect = which_side_effect + mock_nix_candidates.return_value = ["/home/test/.nix-profile/bin/pkgmgr"] + + repo = {"repository": "pkgmgr"} + cmd = resolve_command_for_repo( + repo=repo, + repo_identifier="pkgmgr", + repo_dir="/tmp/pkgmgr", + ) + self.assertEqual(cmd, "/home/test/.nix-profile/bin/pkgmgr") + + def test_main_sh_fallback_when_no_binaries(self) -> None: + """ + If no CLI is found via PATH or Nix, resolve_command_for_repo() + should fall back to an executable main.sh in the repo root. + """ + with tempfile.TemporaryDirectory() as tmpdir, patch( + "pkgmgr.core.command.resolve.shutil.which", return_value=None + ), patch( + "pkgmgr.core.command.resolve._nix_binary_candidates", return_value=[] + ), patch( + "pkgmgr.core.command.resolve._is_executable" + ) as mock_is_executable: + main_sh = os.path.join(tmpdir, "main.sh") + with open(main_sh, "w", encoding="utf-8") as f: + f.write("#!/bin/sh\nexit 0\n") + os.chmod(main_sh, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR) + + def is_exec_side_effect(path: str) -> bool: + return path == main_sh + + mock_is_executable.side_effect = is_exec_side_effect + + repo = {} + cmd = resolve_command_for_repo( + repo=repo, + repo_identifier="pkgmgr", + repo_dir=tmpdir, + ) + self.assertEqual(cmd, main_sh) + + def test_python_package_without_entry_point_returns_none(self) -> None: + """ + If the repository looks like a Python package (src/package/__init__.py) + but there is no CLI entry point or main.sh/main.py, the result + should be None. + """ + with tempfile.TemporaryDirectory() as tmpdir, patch( + "pkgmgr.core.command.resolve.shutil.which", return_value=None + ), patch( + "pkgmgr.core.command.resolve._nix_binary_candidates", return_value=[] + ), patch( + "pkgmgr.core.command.resolve._is_executable", return_value=False + ): + src_dir = os.path.join(tmpdir, "src", "mypkg") + os.makedirs(src_dir, exist_ok=True) + init_path = os.path.join(src_dir, "__init__.py") + with open(init_path, "w", encoding="utf-8") as f: + f.write("# package marker\n") + + repo = {} + cmd = resolve_command_for_repo( + repo=repo, + repo_identifier="mypkg", + repo_dir=tmpdir, + ) + self.assertIsNone(cmd) + + +if __name__ == "__main__": + unittest.main()