diff --git a/tests/unit/pkgmgr/actions/install/installers/nix/_fakes.py b/tests/unit/pkgmgr/actions/install/installers/nix/_fakes.py new file mode 100644 index 0000000..025ca01 --- /dev/null +++ b/tests/unit/pkgmgr/actions/install/installers/nix/_fakes.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Optional + + +@dataclass +class FakeRunResult: + returncode: int + stdout: str = "" + stderr: str = "" + + +class FakeRunner: + """ + Minimal runner stub compatible with: + - CommandRunner.run(ctx, cmd, allow_failure=...) + - Generic runner.run(ctx, cmd, allow_failure=...) + """ + + def __init__(self, mapping: Optional[dict[str, Any]] = None, default: Any = None): + self.mapping = mapping or {} + self.default = default if default is not None else FakeRunResult(0, "", "") + self.calls: list[tuple[Any, str, bool]] = [] + + def run(self, ctx, cmd: str, allow_failure: bool = False): + self.calls.append((ctx, cmd, allow_failure)) + return self.mapping.get(cmd, self.default) + + +class FakeRetry: + """ + Mimics GitHubRateLimitRetry.run_with_retry(ctx, runner, cmd) + """ + + def __init__(self, results: list[FakeRunResult]): + self._results = list(results) + self.calls: list[str] = [] + + def run_with_retry(self, ctx, runner, cmd: str): + self.calls.append(cmd) + if self._results: + return self._results.pop(0) + return FakeRunResult(0, "", "") diff --git a/tests/unit/pkgmgr/actions/install/installers/nix/test_conflicts_resolver.py b/tests/unit/pkgmgr/actions/install/installers/nix/test_conflicts_resolver.py new file mode 100644 index 0000000..830570f --- /dev/null +++ b/tests/unit/pkgmgr/actions/install/installers/nix/test_conflicts_resolver.py @@ -0,0 +1,58 @@ +from __future__ import annotations + +import unittest + +from pkgmgr.actions.install.installers.nix.conflicts import NixConflictResolver +from ._fakes import FakeRunResult, FakeRunner, FakeRetry + + +class DummyCtx: + quiet = True + + +class TestNixConflictResolver(unittest.TestCase): + def test_resolve_removes_tokens_and_retries_success(self) -> None: + ctx = DummyCtx() + install_cmd = "nix profile install /repo#default" + + stderr = ''' + error: An existing package already provides the following file: + /nix/store/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-pkgmgr/bin/pkgmgr + ''' + + runner = FakeRunner(mapping={ + "nix profile remove pkgmgr": FakeRunResult(0, "", ""), + }) + retry = FakeRetry(results=[FakeRunResult(0, "", "")]) + + class FakeProfile: + def find_remove_tokens_for_store_prefixes(self, ctx, runner, prefixes): + return [] + def find_remove_tokens_for_output(self, ctx, runner, output): + return ["pkgmgr"] + + resolver = NixConflictResolver(runner=runner, retry=retry, profile=FakeProfile()) + ok = resolver.resolve(ctx, install_cmd, stdout="", stderr=stderr, output="pkgmgr", max_rounds=2) + self.assertTrue(ok) + self.assertIn("nix profile remove pkgmgr", [c[1] for c in runner.calls]) + + def test_resolve_uses_textual_remove_tokens_last_resort(self) -> None: + ctx = DummyCtx() + install_cmd = "nix profile install /repo#default" + + stderr = "hint: try:\n nix profile remove 'pkgmgr-1'\n" + runner = FakeRunner(mapping={ + "nix profile remove pkgmgr-1": FakeRunResult(0, "", ""), + }) + retry = FakeRetry(results=[FakeRunResult(0, "", "")]) + + class FakeProfile: + def find_remove_tokens_for_store_prefixes(self, ctx, runner, prefixes): + return [] + def find_remove_tokens_for_output(self, ctx, runner, output): + return [] + + resolver = NixConflictResolver(runner=runner, retry=retry, profile=FakeProfile()) + ok = resolver.resolve(ctx, install_cmd, stdout="", stderr=stderr, output="pkgmgr", max_rounds=2) + self.assertTrue(ok) + self.assertIn("nix profile remove pkgmgr-1", [c[1] for c in runner.calls]) diff --git a/tests/unit/pkgmgr/actions/install/installers/nix/test_inspector.py b/tests/unit/pkgmgr/actions/install/installers/nix/test_inspector.py new file mode 100644 index 0000000..1241461 --- /dev/null +++ b/tests/unit/pkgmgr/actions/install/installers/nix/test_inspector.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +import json +import unittest + +from pkgmgr.actions.install.installers.nix.profile import NixProfileInspector +from ._fakes import FakeRunResult, FakeRunner + + +class TestNixProfileInspector(unittest.TestCase): + def test_list_json_accepts_raw_string(self) -> None: + payload = {"elements": {"pkgmgr-1": {"attrPath": "packages.x86_64-linux.pkgmgr"}}} + raw = json.dumps(payload) + runner = FakeRunner(default=raw) + insp = NixProfileInspector() + data = insp.list_json(ctx=None, runner=runner) + self.assertEqual(data["elements"]["pkgmgr-1"]["attrPath"], "packages.x86_64-linux.pkgmgr") + + def test_list_json_accepts_result_object(self) -> None: + payload = {"elements": {"pkgmgr-1": {"attrPath": "packages.x86_64-linux.pkgmgr"}}} + raw = json.dumps(payload) + runner = FakeRunner(default=FakeRunResult(0, stdout=raw)) + insp = NixProfileInspector() + data = insp.list_json(ctx=None, runner=runner) + self.assertEqual(data["elements"]["pkgmgr-1"]["attrPath"], "packages.x86_64-linux.pkgmgr") + + def test_find_remove_tokens_for_output_includes_output_first(self) -> None: + payload = { + "elements": { + "pkgmgr-1": {"name": "pkgmgr-1", "attrPath": "packages.x86_64-linux.pkgmgr"}, + "default-1": {"name": "default-1", "attrPath": "packages.x86_64-linux.default"}, + } + } + raw = json.dumps(payload) + runner = FakeRunner(default=FakeRunResult(0, stdout=raw)) + insp = NixProfileInspector() + tokens = insp.find_remove_tokens_for_output(ctx=None, runner=runner, output="pkgmgr") + self.assertEqual(tokens[0], "pkgmgr") + self.assertIn("pkgmgr-1", tokens) + + def test_find_remove_tokens_for_store_prefixes(self) -> None: + payload = { + "elements": { + "pkgmgr-1": { + "name": "pkgmgr-1", + "attrPath": "packages.x86_64-linux.pkgmgr", + "storePaths": ["/nix/store/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-pkgmgr"], + }, + "something": { + "name": "other", + "attrPath": "packages.x86_64-linux.other", + "storePaths": ["/nix/store/bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb-other"], + }, + } + } + raw = json.dumps(payload) + runner = FakeRunner(default=FakeRunResult(0, stdout=raw)) + insp = NixProfileInspector() + tokens = insp.find_remove_tokens_for_store_prefixes( + ctx=None, runner=runner, prefixes=["/nix/store/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-pkgmgr"] + ) + self.assertIn("pkgmgr-1", tokens) diff --git a/tests/unit/pkgmgr/actions/install/installers/nix/test_installer_core.py b/tests/unit/pkgmgr/actions/install/installers/nix/test_installer_core.py new file mode 100644 index 0000000..6abe65a --- /dev/null +++ b/tests/unit/pkgmgr/actions/install/installers/nix/test_installer_core.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +import unittest +from unittest.mock import MagicMock + +from pkgmgr.actions.install.installers.nix.installer import NixFlakeInstaller +from ._fakes import FakeRunResult + + +class DummyCtx: + def __init__(self, identifier: str = "x", repo_dir: str = "/repo", quiet: bool = True, force_update: bool = False): + self.identifier = identifier + self.repo_dir = repo_dir + self.quiet = quiet + self.force_update = force_update + + +class TestNixFlakeInstallerCore(unittest.TestCase): + def test_install_only_success_returns(self) -> None: + ins = NixFlakeInstaller() + ins.supports = MagicMock(return_value=True) + + ins._retry = MagicMock() + ins._retry.run_with_retry.return_value = FakeRunResult(0, "", "") + ins._conflicts = MagicMock() + ins._profile = MagicMock() + ins._runner = MagicMock() + + ctx = DummyCtx(identifier="lib", repo_dir="/repo", quiet=True) + ins.run(ctx) + ins._retry.run_with_retry.assert_called() + + def test_conflict_resolver_success_short_circuits(self) -> None: + ins = NixFlakeInstaller() + ins.supports = MagicMock(return_value=True) + + ins._retry = MagicMock() + ins._retry.run_with_retry.return_value = FakeRunResult(1, "out", "err") + ins._conflicts = MagicMock() + ins._conflicts.resolve.return_value = True + ins._profile = MagicMock() + ins._runner = MagicMock() + + ctx = DummyCtx(identifier="lib", repo_dir="/repo", quiet=True) + ins.run(ctx) + ins._conflicts.resolve.assert_called() + + def test_mandatory_failure_raises_systemexit(self) -> None: + ins = NixFlakeInstaller() + ins.supports = MagicMock(return_value=True) + + ins._retry = MagicMock() + ins._retry.run_with_retry.return_value = FakeRunResult(2, "", "no") + ins._conflicts = MagicMock() + ins._conflicts.resolve.return_value = False + ins._profile = MagicMock() + ins._profile.find_installed_indices_for_output.return_value = [] + ins._runner = MagicMock() + ins._runner.run.return_value = FakeRunResult(2, "", "") + + ctx = DummyCtx(identifier="lib", repo_dir="/repo", quiet=True) + with self.assertRaises(SystemExit) as cm: + ins.run(ctx) + self.assertEqual(cm.exception.code, 2) + + def test_optional_failure_does_not_raise(self) -> None: + ins = NixFlakeInstaller() + ins.supports = MagicMock(return_value=True) + + results = [ + FakeRunResult(0, "", ""), + FakeRunResult(2, "", ""), + ] + + def run_with_retry(ctx, runner, cmd): + return results.pop(0) + + ins._retry = MagicMock() + ins._retry.run_with_retry.side_effect = run_with_retry + ins._conflicts = MagicMock() + ins._conflicts.resolve.return_value = False + ins._profile = MagicMock() + ins._profile.find_installed_indices_for_output.return_value = [] + ins._runner = MagicMock() + ins._runner.run.return_value = FakeRunResult(2, "", "") + + ctx = DummyCtx(identifier="pkgmgr", repo_dir="/repo", quiet=True) + ins.run(ctx) # must not raise diff --git a/tests/unit/pkgmgr/actions/install/installers/nix/test_matcher.py b/tests/unit/pkgmgr/actions/install/installers/nix/test_matcher.py new file mode 100644 index 0000000..5070653 --- /dev/null +++ b/tests/unit/pkgmgr/actions/install/installers/nix/test_matcher.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +import unittest + +from pkgmgr.actions.install.installers.nix.profile.models import NixProfileEntry +from pkgmgr.actions.install.installers.nix.profile.matcher import entry_matches_output, entry_matches_store_path + + +class TestMatcher(unittest.TestCase): + def _e(self, name: str, attr: str) -> NixProfileEntry: + return NixProfileEntry( + key="pkgmgr-1", + index=None, + name=name, + attr_path=attr, + store_paths=["/nix/store/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-pkgmgr"], + ) + + def test_matches_direct_name(self) -> None: + self.assertTrue(entry_matches_output(self._e("pkgmgr", ""), "pkgmgr")) + + def test_matches_attrpath_hash(self) -> None: + self.assertTrue(entry_matches_output(self._e("", "github:me/repo#pkgmgr"), "pkgmgr")) + + def test_matches_attrpath_dot_suffix(self) -> None: + self.assertTrue(entry_matches_output(self._e("", "packages.x86_64-linux.pkgmgr"), "pkgmgr")) + + def test_matches_name_with_suffix_number(self) -> None: + self.assertTrue(entry_matches_output(self._e("pkgmgr-1", ""), "pkgmgr")) + + def test_package_manager_special_case(self) -> None: + self.assertTrue(entry_matches_output(self._e("package-manager-2", ""), "pkgmgr")) + + def test_store_path_match(self) -> None: + entry = self._e("pkgmgr-1", "") + self.assertTrue(entry_matches_store_path(entry, "/nix/store/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-pkgmgr")) + self.assertFalse(entry_matches_store_path(entry, "/nix/store/cccccccccccccccccccccccccccccccc-zzz")) diff --git a/tests/unit/pkgmgr/actions/install/installers/nix/test_normalizer.py b/tests/unit/pkgmgr/actions/install/installers/nix/test_normalizer.py new file mode 100644 index 0000000..f873774 --- /dev/null +++ b/tests/unit/pkgmgr/actions/install/installers/nix/test_normalizer.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +import unittest + +from pkgmgr.actions.install.installers.nix.profile.normalizer import coerce_index, normalize_elements + + +class TestNormalizer(unittest.TestCase): + def test_coerce_index_numeric_key(self) -> None: + self.assertEqual(coerce_index("3", {"name": "x"}), 3) + + def test_coerce_index_explicit_field(self) -> None: + self.assertEqual(coerce_index("pkgmgr-1", {"index": 7}), 7) + self.assertEqual(coerce_index("pkgmgr-1", {"id": "8"}), 8) + + def test_coerce_index_trailing_number(self) -> None: + self.assertEqual(coerce_index("pkgmgr-42", {"name": "x"}), 42) + + def test_normalize_elements_handles_missing_elements(self) -> None: + self.assertEqual(normalize_elements({}), []) + + def test_normalize_elements_collects_store_paths(self) -> None: + data = { + "elements": { + "pkgmgr-1": { + "name": "pkgmgr-1", + "attrPath": "packages.x86_64-linux.pkgmgr", + "storePaths": ["/nix/store/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-pkgmgr"], + }, + "2": { + "name": "foo", + "attrPath": "packages.x86_64-linux.default", + "storePath": "/nix/store/bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb-foo", + }, + } + } + entries = normalize_elements(data) + self.assertEqual(len(entries), 2) + self.assertTrue(entries[0].store_paths) diff --git a/tests/unit/pkgmgr/actions/install/installers/nix/test_parser.py b/tests/unit/pkgmgr/actions/install/installers/nix/test_parser.py new file mode 100644 index 0000000..a8b2f69 --- /dev/null +++ b/tests/unit/pkgmgr/actions/install/installers/nix/test_parser.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +import json +import unittest + +from pkgmgr.actions.install.installers.nix.profile.parser import parse_profile_list_json + + +class TestParseProfileListJson(unittest.TestCase): + def test_parses_valid_json(self) -> None: + payload = {"elements": {"0": {"name": "pkgmgr"}}} + raw = json.dumps(payload) + self.assertEqual(parse_profile_list_json(raw)["elements"]["0"]["name"], "pkgmgr") + + def test_raises_systemexit_on_invalid_json(self) -> None: + with self.assertRaises(SystemExit) as cm: + parse_profile_list_json("{not json") + self.assertIn("Failed to parse", str(cm.exception)) diff --git a/tests/unit/pkgmgr/actions/install/installers/nix/test_profile_list_reader.py b/tests/unit/pkgmgr/actions/install/installers/nix/test_profile_list_reader.py new file mode 100644 index 0000000..f961c1d --- /dev/null +++ b/tests/unit/pkgmgr/actions/install/installers/nix/test_profile_list_reader.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +import unittest + +from pkgmgr.actions.install.installers.nix.profile_list import NixProfileListReader +from ._fakes import FakeRunResult, FakeRunner + + +class TestNixProfileListReader(unittest.TestCase): + def test_entries_parses_indices_and_store_prefixes(self) -> None: + out = ''' + 0 something /nix/store/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-pkgmgr + 1 something /nix/store/bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb-foo + ''' + runner = FakeRunner(default=FakeRunResult(0, stdout=out)) + reader = NixProfileListReader(runner=runner) + entries = reader.entries(ctx=None) + self.assertEqual(entries[0][0], 0) + self.assertTrue(entries[0][1].startswith("/nix/store/")) + + def test_indices_matching_store_prefixes(self) -> None: + out = " 7 x /nix/store/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-pkgmgr\n" + runner = FakeRunner(default=FakeRunResult(0, stdout=out)) + reader = NixProfileListReader(runner=runner) + hits = reader.indices_matching_store_prefixes( + ctx=None, + prefixes=["/nix/store/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-pkgmgr"], + ) + self.assertEqual(hits, [7]) diff --git a/tests/unit/pkgmgr/actions/install/installers/nix/test_result.py b/tests/unit/pkgmgr/actions/install/installers/nix/test_result.py new file mode 100644 index 0000000..0fca68f --- /dev/null +++ b/tests/unit/pkgmgr/actions/install/installers/nix/test_result.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +import unittest + +from pkgmgr.actions.install.installers.nix.profile.result import extract_stdout_text + + +class TestExtractStdoutText(unittest.TestCase): + def test_accepts_string(self) -> None: + self.assertEqual(extract_stdout_text("hello"), "hello") + + def test_accepts_bytes(self) -> None: + self.assertEqual(extract_stdout_text(b"hi"), "hi") + + def test_accepts_object_with_stdout_str(self) -> None: + class R: + stdout = "ok" + self.assertEqual(extract_stdout_text(R()), "ok") + + def test_accepts_object_with_stdout_bytes(self) -> None: + class R: + stdout = b"ok" + self.assertEqual(extract_stdout_text(R()), "ok") + + def test_fallback_str(self) -> None: + class R: + def __str__(self) -> str: + return "repr" + self.assertEqual(extract_stdout_text(R()), "repr") diff --git a/tests/unit/pkgmgr/actions/install/installers/nix/test_textparse.py b/tests/unit/pkgmgr/actions/install/installers/nix/test_textparse.py new file mode 100644 index 0000000..e509362 --- /dev/null +++ b/tests/unit/pkgmgr/actions/install/installers/nix/test_textparse.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +import unittest + +from pkgmgr.actions.install.installers.nix.textparse import NixConflictTextParser + + +class TestNixConflictTextParser(unittest.TestCase): + def test_remove_tokens_parses_unquoted_and_quoted(self) -> None: + t = NixConflictTextParser() + text = ''' + nix profile remove pkgmgr + nix profile remove 'pkgmgr-1' + nix profile remove "default-2" + ''' + tokens = t.remove_tokens(text) + self.assertEqual(tokens, ["pkgmgr", "pkgmgr-1", "default-2"]) + + def test_existing_store_prefixes_extracts_existing_section_only(self) -> None: + t = NixConflictTextParser() + text = ''' + error: An existing package already provides the following file: + /nix/store/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-pkgmgr/bin/pkgmgr + /nix/store/bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb-pkgmgr/share/doc + This is the conflicting file from the new package: + /nix/store/cccccccccccccccccccccccccccccccc-pkgmgr/bin/pkgmgr + ''' + prefixes = t.existing_store_prefixes(text) + self.assertEqual(len(prefixes), 2) + self.assertTrue(prefixes[0].startswith("/nix/store/"))