fix(nix): resolve Ruff F821 via TYPE_CHECKING and stabilize NixFlakeInstaller tests
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 / codesniffer-shellcheck (push) Has been cancelled
Mark stable commit / codesniffer-ruff (push) Has been cancelled
Mark stable commit / mark-stable (push) Has been cancelled
CI / test-unit (push) Has been cancelled
CI / test-integration (push) Has been cancelled
CI / test-env-virtual (push) Has been cancelled
CI / test-env-nix (push) Has been cancelled
CI / test-e2e (push) Has been cancelled
CI / test-virgin-user (push) Has been cancelled
CI / test-virgin-root (push) Has been cancelled
CI / codesniffer-shellcheck (push) Has been cancelled
CI / codesniffer-ruff (push) Has been cancelled
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 / codesniffer-shellcheck (push) Has been cancelled
Mark stable commit / codesniffer-ruff (push) Has been cancelled
Mark stable commit / mark-stable (push) Has been cancelled
CI / test-unit (push) Has been cancelled
CI / test-integration (push) Has been cancelled
CI / test-env-virtual (push) Has been cancelled
CI / test-env-nix (push) Has been cancelled
CI / test-e2e (push) Has been cancelled
CI / test-virgin-user (push) Has been cancelled
CI / test-virgin-root (push) Has been cancelled
CI / codesniffer-shellcheck (push) Has been cancelled
CI / codesniffer-ruff (push) Has been cancelled
- Add TYPE_CHECKING imports for RepoContext and CommandRunner to avoid runtime deps - Fix Ruff F821 undefined-name errors in nix installer modules - Refactor legacy NixFlakeInstaller unit tests to mock subprocess.run directly - Remove obsolete run_cmd_mock usage and assert install calls via subprocess calls - Ensure tests run without realtime waits or external nix dependencies https://chatgpt.com/share/693e925d-a79c-800f-b0b6-92b8ba260b11
This commit is contained in:
@@ -3,7 +3,7 @@ from __future__ import annotations
|
|||||||
|
|
||||||
import os
|
import os
|
||||||
import shutil
|
import shutil
|
||||||
from typing import List, Tuple
|
from typing import List, Tuple, TYPE_CHECKING
|
||||||
|
|
||||||
from pkgmgr.actions.install.installers.base import BaseInstaller
|
from pkgmgr.actions.install.installers.base import BaseInstaller
|
||||||
|
|
||||||
@@ -11,6 +11,8 @@ from .profile import NixProfileInspector
|
|||||||
from .retry import GitHubRateLimitRetry, RetryPolicy
|
from .retry import GitHubRateLimitRetry, RetryPolicy
|
||||||
from .runner import CommandRunner
|
from .runner import CommandRunner
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
from pkgmgr.actions.install.context import RepoContext
|
||||||
|
|
||||||
class NixFlakeInstaller(BaseInstaller):
|
class NixFlakeInstaller(BaseInstaller):
|
||||||
layer = "nix"
|
layer = "nix"
|
||||||
|
|||||||
@@ -1,8 +1,12 @@
|
|||||||
# src/pkgmgr/actions/install/installers/nix/profile.py
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import json
|
import json
|
||||||
from typing import Any, List
|
from typing import Any, List, TYPE_CHECKING
|
||||||
|
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
from pkgmgr.actions.install.context import RepoContext
|
||||||
|
from .runner import CommandRunner
|
||||||
|
|
||||||
class NixProfileInspector:
|
class NixProfileInspector:
|
||||||
"""
|
"""
|
||||||
|
|||||||
@@ -1,13 +1,15 @@
|
|||||||
# src/pkgmgr/actions/install/installers/nix/retry.py
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import random
|
import random
|
||||||
import time
|
import time
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from typing import Iterable
|
from typing import Iterable, TYPE_CHECKING
|
||||||
|
|
||||||
from .types import RunResult
|
from .types import RunResult
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
from pkgmgr.actions.install.context import RepoContext
|
||||||
|
from .runner import CommandRunner
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
class RetryPolicy:
|
class RetryPolicy:
|
||||||
|
|||||||
@@ -2,8 +2,12 @@ from __future__ import annotations
|
|||||||
|
|
||||||
import subprocess
|
import subprocess
|
||||||
|
|
||||||
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
from .types import RunResult
|
from .types import RunResult
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
from pkgmgr.actions.install.context import RepoContext
|
||||||
|
|
||||||
class CommandRunner:
|
class CommandRunner:
|
||||||
"""
|
"""
|
||||||
|
|||||||
@@ -20,6 +20,7 @@ import subprocess
|
|||||||
import tempfile
|
import tempfile
|
||||||
import unittest
|
import unittest
|
||||||
from contextlib import redirect_stdout
|
from contextlib import redirect_stdout
|
||||||
|
from typing import List
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
from pkgmgr.actions.install.installers.nix import NixFlakeInstaller
|
from pkgmgr.actions.install.installers.nix import NixFlakeInstaller
|
||||||
@@ -60,42 +61,51 @@ class TestNixFlakeInstaller(unittest.TestCase):
|
|||||||
shutil.rmtree(self._tmpdir, ignore_errors=True)
|
shutil.rmtree(self._tmpdir, ignore_errors=True)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _cp(code: int) -> subprocess.CompletedProcess:
|
def _cp(code: int, stdout: str = "", stderr: str = "") -> subprocess.CompletedProcess:
|
||||||
# stdout/stderr are irrelevant here, but keep shape realistic
|
return subprocess.CompletedProcess(args=["nix"], returncode=code, stdout=stdout, stderr=stderr)
|
||||||
return subprocess.CompletedProcess(args=["nix"], returncode=code, stdout="", stderr="")
|
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _enable_nix_in_module(which_patch) -> None:
|
def _enable_nix_in_module(which_patch) -> None:
|
||||||
"""Ensure shutil.which('nix') in nix module returns a path."""
|
"""Ensure shutil.which('nix') in nix installer module returns a path."""
|
||||||
which_patch.return_value = "/usr/bin/nix"
|
which_patch.return_value = "/usr/bin/nix"
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _install_cmds_from_calls(call_args_list) -> List[str]:
|
||||||
|
cmds: List[str] = []
|
||||||
|
for c in call_args_list:
|
||||||
|
if not c.args:
|
||||||
|
continue
|
||||||
|
cmd = c.args[0]
|
||||||
|
if isinstance(cmd, str) and cmd.startswith("nix profile install "):
|
||||||
|
cmds.append(cmd)
|
||||||
|
return cmds
|
||||||
|
|
||||||
def test_nix_flake_run_success(self) -> None:
|
def test_nix_flake_run_success(self) -> None:
|
||||||
"""
|
"""
|
||||||
When run_command returns success (returncode 0), installer
|
When install returns success (returncode 0), installer
|
||||||
should report success and not raise.
|
should report success and not raise.
|
||||||
"""
|
"""
|
||||||
ctx = DummyCtx(identifier="some-lib", repo_dir=self.repo_dir)
|
ctx = DummyCtx(identifier="some-lib", repo_dir=self.repo_dir)
|
||||||
installer = NixFlakeInstaller()
|
installer = NixFlakeInstaller()
|
||||||
|
|
||||||
|
install_results = [self._cp(0)] # first install succeeds
|
||||||
|
|
||||||
|
def fake_subprocess_run(cmd, *args, **kwargs):
|
||||||
|
# cmd is a string because CommandRunner uses shell=True
|
||||||
|
if isinstance(cmd, str) and cmd.startswith("nix profile list --json"):
|
||||||
|
return self._cp(0, stdout='{"elements": []}', stderr="")
|
||||||
|
if isinstance(cmd, str) and cmd.startswith("nix profile install "):
|
||||||
|
return install_results.pop(0)
|
||||||
|
return self._cp(0)
|
||||||
|
|
||||||
buf = io.StringIO()
|
buf = io.StringIO()
|
||||||
with patch("pkgmgr.actions.install.installers.nix.installer.shutil.which") as which_mock, patch(
|
with patch("pkgmgr.actions.install.installers.nix.installer.shutil.which") as which_mock, patch(
|
||||||
"pkgmgr.actions.install.installers.nix.installer.os.path.exists", return_value=True
|
"pkgmgr.actions.install.installers.nix.installer.os.path.exists", return_value=True
|
||||||
), patch(
|
), patch(
|
||||||
"pkgmgr.actions.install.installers.nix.runner.subprocess.run"
|
"pkgmgr.actions.install.installers.nix.runner.subprocess.run", side_effect=fake_subprocess_run
|
||||||
) as subproc_mock, redirect_stdout(buf):
|
) as subproc_mock, redirect_stdout(buf):
|
||||||
|
|
||||||
self._enable_nix_in_module(which_mock)
|
self._enable_nix_in_module(which_mock)
|
||||||
|
|
||||||
subproc_mock.return_value = subprocess.CompletedProcess(
|
|
||||||
args=["nix", "profile", "list", "--json"],
|
|
||||||
returncode=0,
|
|
||||||
stdout='{"elements": []}',
|
|
||||||
stderr="",
|
|
||||||
)
|
|
||||||
|
|
||||||
# Install succeeds
|
|
||||||
run_cmd_mock.return_value = self._cp(0)
|
|
||||||
|
|
||||||
self.assertTrue(installer.supports(ctx))
|
self.assertTrue(installer.supports(ctx))
|
||||||
installer.run(ctx)
|
installer.run(ctx)
|
||||||
|
|
||||||
@@ -103,12 +113,8 @@ class TestNixFlakeInstaller(unittest.TestCase):
|
|||||||
self.assertIn("[nix] install: nix profile install", out)
|
self.assertIn("[nix] install: nix profile install", out)
|
||||||
self.assertIn("[nix] output 'default' successfully installed.", out)
|
self.assertIn("[nix] output 'default' successfully installed.", out)
|
||||||
|
|
||||||
run_cmd_mock.assert_called_with(
|
install_cmds = self._install_cmds_from_calls(subproc_mock.call_args_list)
|
||||||
f"nix profile install {self.repo_dir}#default",
|
self.assertEqual(install_cmds, [f"nix profile install {self.repo_dir}#default"])
|
||||||
cwd=self.repo_dir,
|
|
||||||
preview=False,
|
|
||||||
allow_failure=True,
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_nix_flake_run_mandatory_failure_raises(self) -> None:
|
def test_nix_flake_run_mandatory_failure_raises(self) -> None:
|
||||||
"""
|
"""
|
||||||
@@ -118,34 +124,43 @@ class TestNixFlakeInstaller(unittest.TestCase):
|
|||||||
ctx = DummyCtx(identifier="some-lib", repo_dir=self.repo_dir)
|
ctx = DummyCtx(identifier="some-lib", repo_dir=self.repo_dir)
|
||||||
installer = NixFlakeInstaller()
|
installer = NixFlakeInstaller()
|
||||||
|
|
||||||
|
# retry layer does one attempt (non-403), then fallback does final attempt => 2 installs
|
||||||
|
install_results = [self._cp(1), self._cp(1)]
|
||||||
|
|
||||||
|
def fake_subprocess_run(cmd, *args, **kwargs):
|
||||||
|
if isinstance(cmd, str) and cmd.startswith("nix profile list --json"):
|
||||||
|
return self._cp(0, stdout='{"elements": []}', stderr="")
|
||||||
|
if isinstance(cmd, str) and cmd.startswith("nix profile install "):
|
||||||
|
return install_results.pop(0)
|
||||||
|
return self._cp(0)
|
||||||
|
|
||||||
buf = io.StringIO()
|
buf = io.StringIO()
|
||||||
with patch("pkgmgr.actions.install.installers.nix.installer.shutil.which") as which_mock, patch(
|
with patch("pkgmgr.actions.install.installers.nix.installer.shutil.which") as which_mock, patch(
|
||||||
"pkgmgr.actions.install.installers.nix.installer.os.path.exists", return_value=True
|
"pkgmgr.actions.install.installers.nix.installer.os.path.exists", return_value=True
|
||||||
), patch(
|
), patch(
|
||||||
"pkgmgr.actions.install.installers.nix.runner.subprocess.run"
|
"pkgmgr.actions.install.installers.nix.runner.subprocess.run", side_effect=fake_subprocess_run
|
||||||
) as subproc_mock, redirect_stdout(buf):
|
) as subproc_mock, redirect_stdout(buf):
|
||||||
|
|
||||||
self._enable_nix_in_module(which_mock)
|
self._enable_nix_in_module(which_mock)
|
||||||
|
|
||||||
subproc_mock.return_value = subprocess.CompletedProcess(
|
|
||||||
args=["nix", "profile", "list", "--json"],
|
|
||||||
returncode=0,
|
|
||||||
stdout='{"elements": []}',
|
|
||||||
stderr="",
|
|
||||||
)
|
|
||||||
|
|
||||||
# First install fails, retry fails -> should raise SystemExit(1)
|
|
||||||
run_cmd_mock.side_effect = [self._cp(1), self._cp(1)]
|
|
||||||
|
|
||||||
self.assertTrue(installer.supports(ctx))
|
self.assertTrue(installer.supports(ctx))
|
||||||
with self.assertRaises(SystemExit) as cm:
|
with self.assertRaises(SystemExit) as cm:
|
||||||
installer.run(ctx)
|
installer.run(ctx)
|
||||||
|
|
||||||
self.assertEqual(cm.exception.code, 1)
|
self.assertEqual(cm.exception.code, 1)
|
||||||
|
|
||||||
out = buf.getvalue()
|
out = buf.getvalue()
|
||||||
self.assertIn("[nix] install: nix profile install", out)
|
self.assertIn("[nix] install: nix profile install", out)
|
||||||
self.assertIn("[ERROR] Failed to install Nix flake output 'default' (exit 1)", out)
|
self.assertIn("[ERROR] Failed to install Nix flake output 'default' (exit 1)", out)
|
||||||
|
|
||||||
|
install_cmds = self._install_cmds_from_calls(subproc_mock.call_args_list)
|
||||||
|
self.assertEqual(
|
||||||
|
install_cmds,
|
||||||
|
[
|
||||||
|
f"nix profile install {self.repo_dir}#default",
|
||||||
|
f"nix profile install {self.repo_dir}#default",
|
||||||
|
],
|
||||||
|
)
|
||||||
|
|
||||||
def test_nix_flake_run_optional_failure_does_not_raise(self) -> None:
|
def test_nix_flake_run_optional_failure_does_not_raise(self) -> None:
|
||||||
"""
|
"""
|
||||||
For pkgmgr/package-manager repositories:
|
For pkgmgr/package-manager repositories:
|
||||||
@@ -156,29 +171,26 @@ class TestNixFlakeInstaller(unittest.TestCase):
|
|||||||
ctx = DummyCtx(identifier="pkgmgr", repo_dir=self.repo_dir)
|
ctx = DummyCtx(identifier="pkgmgr", repo_dir=self.repo_dir)
|
||||||
installer = NixFlakeInstaller()
|
installer = NixFlakeInstaller()
|
||||||
|
|
||||||
|
# pkgmgr success (1 call), default fails (2 calls: attempt + final)
|
||||||
|
install_results = [self._cp(0), self._cp(1), self._cp(1)]
|
||||||
|
|
||||||
|
def fake_subprocess_run(cmd, *args, **kwargs):
|
||||||
|
if isinstance(cmd, str) and cmd.startswith("nix profile list --json"):
|
||||||
|
return self._cp(0, stdout='{"elements": []}', stderr="")
|
||||||
|
if isinstance(cmd, str) and cmd.startswith("nix profile install "):
|
||||||
|
return install_results.pop(0)
|
||||||
|
return self._cp(0)
|
||||||
|
|
||||||
buf = io.StringIO()
|
buf = io.StringIO()
|
||||||
with patch("pkgmgr.actions.install.installers.nix.installer.shutil.which") as which_mock, patch(
|
with patch("pkgmgr.actions.install.installers.nix.installer.shutil.which") as which_mock, patch(
|
||||||
"pkgmgr.actions.install.installers.nix.installer.os.path.exists", return_value=True
|
"pkgmgr.actions.install.installers.nix.installer.os.path.exists", return_value=True
|
||||||
), patch(
|
), patch(
|
||||||
"pkgmgr.actions.install.installers.nix.runner.subprocess.run"
|
"pkgmgr.actions.install.installers.nix.runner.subprocess.run", side_effect=fake_subprocess_run
|
||||||
) as subproc_mock, redirect_stdout(buf):
|
) as subproc_mock, redirect_stdout(buf):
|
||||||
|
|
||||||
self._enable_nix_in_module(which_mock)
|
self._enable_nix_in_module(which_mock)
|
||||||
|
|
||||||
subproc_mock.return_value = subprocess.CompletedProcess(
|
|
||||||
args=["nix", "profile", "list", "--json"],
|
|
||||||
returncode=0,
|
|
||||||
stdout='{"elements": []}',
|
|
||||||
stderr="",
|
|
||||||
)
|
|
||||||
|
|
||||||
# pkgmgr install ok; default fails twice (initial + retry)
|
|
||||||
run_cmd_mock.side_effect = [self._cp(0), self._cp(1), self._cp(1)]
|
|
||||||
|
|
||||||
self.assertTrue(installer.supports(ctx))
|
self.assertTrue(installer.supports(ctx))
|
||||||
|
installer.run(ctx) # must NOT raise
|
||||||
# Must NOT raise despite optional failure
|
|
||||||
installer.run(ctx)
|
|
||||||
|
|
||||||
out = buf.getvalue()
|
out = buf.getvalue()
|
||||||
|
|
||||||
@@ -192,14 +204,15 @@ class TestNixFlakeInstaller(unittest.TestCase):
|
|||||||
self.assertIn("[ERROR] Failed to install Nix flake output 'default' (exit 1)", out)
|
self.assertIn("[ERROR] Failed to install Nix flake output 'default' (exit 1)", out)
|
||||||
self.assertIn("[WARNING] Continuing despite failure of optional output 'default'.", out)
|
self.assertIn("[WARNING] Continuing despite failure of optional output 'default'.", out)
|
||||||
|
|
||||||
# Verify run_command was called for both outputs (default twice due to retry)
|
install_cmds = self._install_cmds_from_calls(subproc_mock.call_args_list)
|
||||||
expected_calls = [
|
self.assertEqual(
|
||||||
(f"nix profile install {self.repo_dir}#pkgmgr",),
|
install_cmds,
|
||||||
(f"nix profile install {self.repo_dir}#default",),
|
[
|
||||||
(f"nix profile install {self.repo_dir}#default",),
|
f"nix profile install {self.repo_dir}#pkgmgr",
|
||||||
]
|
f"nix profile install {self.repo_dir}#default",
|
||||||
actual_cmds = [c.args[0] for c in run_cmd_mock.call_args_list]
|
f"nix profile install {self.repo_dir}#default",
|
||||||
self.assertEqual(actual_cmds, [e[0] for e in expected_calls])
|
],
|
||||||
|
)
|
||||||
|
|
||||||
def test_nix_flake_supports_respects_disable_env(self) -> None:
|
def test_nix_flake_supports_respects_disable_env(self) -> None:
|
||||||
"""
|
"""
|
||||||
@@ -216,5 +229,6 @@ class TestNixFlakeInstaller(unittest.TestCase):
|
|||||||
os.environ["PKGMGR_DISABLE_NIX_FLAKE_INSTALLER"] = "1"
|
os.environ["PKGMGR_DISABLE_NIX_FLAKE_INSTALLER"] = "1"
|
||||||
self.assertFalse(installer.supports(ctx))
|
self.assertFalse(installer.supports(ctx))
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|||||||
Reference in New Issue
Block a user