**test(nix): add comprehensive unittest coverage for nix installer helpers**
Some checks failed
Mark stable commit / test-unit (push) Has been cancelled
Mark stable commit / test-integration (push) Has been cancelled
Mark stable commit / test-env-virtual (push) Has been cancelled
Mark stable commit / test-env-nix (push) Has been cancelled
Mark stable commit / test-e2e (push) Has been cancelled
Mark stable commit / test-virgin-user (push) Has been cancelled
Mark stable commit / test-virgin-root (push) Has been cancelled
Mark stable commit / linter-shell (push) Has been cancelled
Mark stable commit / linter-python (push) Has been cancelled
Mark stable commit / mark-stable (push) Has been cancelled

* Add reusable fakes for runner and retry logic
* Cover conflict resolution paths (store-prefix, output-token, textual fallback)
* Add unit tests for profile parsing, normalization, matching, and text parsing
* Verify installer core behavior for success, mandatory failure, and optional failure
* Keep tests Nix-free using pure unittest + mocks

https://chatgpt.com/share/693efe80-d928-800f-98b7-0aaafee1d32a
This commit is contained in:
Kevin Veen-Birkenbach
2025-12-14 19:27:26 +01:00
parent ac16378807
commit 328203ccd7
10 changed files with 434 additions and 0 deletions

View File

@@ -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, "", "")

View File

@@ -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])

View File

@@ -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)

View File

@@ -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

View File

@@ -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"))

View File

@@ -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)

View File

@@ -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))

View File

@@ -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])

View File

@@ -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")

View File

@@ -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/"))