executet 'ruff format --check .'
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 / lint-shell (push) Has been cancelled
Mark stable commit / lint-python (push) Has been cancelled
Mark stable commit / mark-stable (push) Has been cancelled

This commit is contained in:
Kevin Veen-Birkenbach
2025-12-18 14:04:44 +01:00
parent 763f02a9a4
commit f4339a746a
155 changed files with 1327 additions and 636 deletions

View File

@@ -227,9 +227,7 @@ class TestBranchCLI(unittest.TestCase):
Ensure that `pkgmgr branch drop <name> --force` passes force=True.
"""
parser = self._create_parser()
args = parser.parse_args(
["branch", "drop", "feature/tmp-branch", "--force"]
)
args = parser.parse_args(["branch", "drop", "feature/tmp-branch", "--force"])
self.assertTrue(args.force)

View File

@@ -83,7 +83,10 @@ class TestInstallReposIntegration(unittest.TestCase):
selected_repos = [repo_system, repo_nix]
all_repos = selected_repos
with tempfile.TemporaryDirectory() as tmp_base, tempfile.TemporaryDirectory() as tmp_bin:
with (
tempfile.TemporaryDirectory() as tmp_base,
tempfile.TemporaryDirectory() as tmp_bin,
):
# Fake repo directories (what get_repo_dir will return)
repo_system_dir = os.path.join(tmp_base, "repo-system")
repo_nix_dir = os.path.join(tmp_base, "repo-nix")
@@ -103,11 +106,12 @@ class TestInstallReposIntegration(unittest.TestCase):
# Patch resolve_command_for_repo at the *pipeline* module level,
# because InstallationPipeline imports it there.
with patch(
"pkgmgr.actions.install.pipeline.resolve_command_for_repo"
) as mock_resolve, patch(
"pkgmgr.actions.install.os.path.exists"
) as mock_exists_install:
with (
patch(
"pkgmgr.actions.install.pipeline.resolve_command_for_repo"
) as mock_resolve,
patch("pkgmgr.actions.install.os.path.exists") as mock_exists_install,
):
def fake_resolve(repo, repo_identifier: str, repo_dir: str):
"""

View File

@@ -29,7 +29,9 @@ from unittest.mock import MagicMock, PropertyMock, patch
class TestIntegrationMirrorCommands(unittest.TestCase):
"""Integration tests for `pkgmgr mirror` commands."""
def _run_pkgmgr(self, args: List[str], extra_env: Optional[Dict[str, str]] = None) -> str:
def _run_pkgmgr(
self, args: List[str], extra_env: Optional[Dict[str, str]] = None
) -> str:
"""Execute pkgmgr with the given arguments and return captured output."""
original_argv = list(sys.argv)
original_env = dict(os.environ)
@@ -80,20 +82,65 @@ class TestIntegrationMirrorCommands(unittest.TestCase):
with ExitStack() as stack:
# build_context is imported directly in these modules:
stack.enter_context(_p("pkgmgr.actions.mirror.list_cmd.build_context", return_value=dummy_ctx))
stack.enter_context(_p("pkgmgr.actions.mirror.diff_cmd.build_context", return_value=dummy_ctx))
stack.enter_context(_p("pkgmgr.actions.mirror.merge_cmd.build_context", return_value=dummy_ctx))
stack.enter_context(_p("pkgmgr.actions.mirror.setup_cmd.build_context", return_value=dummy_ctx))
stack.enter_context(_p("pkgmgr.actions.mirror.remote_provision.build_context", return_value=dummy_ctx))
stack.enter_context(
_p(
"pkgmgr.actions.mirror.list_cmd.build_context",
return_value=dummy_ctx,
)
)
stack.enter_context(
_p(
"pkgmgr.actions.mirror.diff_cmd.build_context",
return_value=dummy_ctx,
)
)
stack.enter_context(
_p(
"pkgmgr.actions.mirror.merge_cmd.build_context",
return_value=dummy_ctx,
)
)
stack.enter_context(
_p(
"pkgmgr.actions.mirror.setup_cmd.build_context",
return_value=dummy_ctx,
)
)
stack.enter_context(
_p(
"pkgmgr.actions.mirror.remote_provision.build_context",
return_value=dummy_ctx,
)
)
# Deterministic remote probing (new refactor: probe_remote_reachable)
stack.enter_context(_p("pkgmgr.core.git.queries.probe_remote_reachable", return_value=True))
stack.enter_context(_p("pkgmgr.actions.mirror.setup_cmd.probe_remote_reachable", return_value=True))
stack.enter_context(
_p(
"pkgmgr.core.git.queries.probe_remote_reachable",
return_value=True,
)
)
stack.enter_context(
_p(
"pkgmgr.actions.mirror.setup_cmd.probe_remote_reachable",
return_value=True,
)
)
# setup_cmd imports ensure_origin_remote directly:
stack.enter_context(_p("pkgmgr.actions.mirror.setup_cmd.ensure_origin_remote", return_value=None))
stack.enter_context(
_p(
"pkgmgr.actions.mirror.setup_cmd.ensure_origin_remote",
return_value=None,
)
)
# Extra safety: if any code calls git_remote.ensure_origin_remote directly
stack.enter_context(_p("pkgmgr.actions.mirror.git_remote.ensure_origin_remote", return_value=None))
stack.enter_context(
_p(
"pkgmgr.actions.mirror.git_remote.ensure_origin_remote",
return_value=None,
)
)
# remote provisioning: remote_provision imports ensure_remote_repo directly from core:
stack.enter_context(
@@ -135,8 +182,12 @@ class TestIntegrationMirrorCommands(unittest.TestCase):
self.assertTrue(output.strip(), "Expected output from mirror diff")
def test_mirror_merge_config_to_file_preview_all(self) -> None:
output = self._run_pkgmgr(["mirror", "merge", "config", "file", "--preview", "--all"])
self.assertTrue(output.strip(), "Expected output from mirror merge (config -> file)")
output = self._run_pkgmgr(
["mirror", "merge", "config", "file", "--preview", "--all"]
)
self.assertTrue(
output.strip(), "Expected output from mirror merge (config -> file)"
)
def test_mirror_setup_preview_all(self) -> None:
output = self._run_pkgmgr(["mirror", "setup", "--preview", "--all"])
@@ -148,7 +199,9 @@ class TestIntegrationMirrorCommands(unittest.TestCase):
def test_mirror_provision_preview_all(self) -> None:
output = self._run_pkgmgr(["mirror", "provision", "--preview", "--all"])
self.assertTrue(output.strip(), "Expected output from mirror provision (preview)")
self.assertTrue(
output.strip(), "Expected output from mirror provision (preview)"
)
if __name__ == "__main__":

View File

@@ -10,6 +10,7 @@ class FakeRunResult:
"""
Mimics your runner returning a structured result object.
"""
returncode: int
stdout: str
stderr: str = ""
@@ -19,6 +20,7 @@ class FakeRunner:
"""
Minimal runner stub: returns exactly what we configure.
"""
def __init__(self, result):
self._result = result
@@ -37,26 +39,34 @@ class TestE2ENixProfileListJsonParsing(unittest.TestCase):
def test_list_json_accepts_raw_string(self) -> None:
from pkgmgr.actions.install.installers.nix.profile import NixProfileInspector
payload = {"elements": {"pkgmgr-1": {"attrPath": "packages.x86_64-linux.pkgmgr"}}}
payload = {
"elements": {"pkgmgr-1": {"attrPath": "packages.x86_64-linux.pkgmgr"}}
}
raw = json.dumps(payload)
runner = FakeRunner(raw)
inspector = NixProfileInspector()
data = inspector.list_json(ctx=None, runner=runner)
self.assertEqual(data["elements"]["pkgmgr-1"]["attrPath"], "packages.x86_64-linux.pkgmgr")
self.assertEqual(
data["elements"]["pkgmgr-1"]["attrPath"], "packages.x86_64-linux.pkgmgr"
)
def test_list_json_accepts_runresult_object(self) -> None:
from pkgmgr.actions.install.installers.nix.profile import NixProfileInspector
payload = {"elements": {"pkgmgr-1": {"attrPath": "packages.x86_64-linux.pkgmgr"}}}
payload = {
"elements": {"pkgmgr-1": {"attrPath": "packages.x86_64-linux.pkgmgr"}}
}
raw = json.dumps(payload)
runner = FakeRunner(FakeRunResult(returncode=0, stdout=raw))
inspector = NixProfileInspector()
data = inspector.list_json(ctx=None, runner=runner)
self.assertEqual(data["elements"]["pkgmgr-1"]["attrPath"], "packages.x86_64-linux.pkgmgr")
self.assertEqual(
data["elements"]["pkgmgr-1"]["attrPath"], "packages.x86_64-linux.pkgmgr"
)
if __name__ == "__main__":

View File

@@ -74,26 +74,28 @@ class TestRecursiveCapabilitiesIntegration(unittest.TestCase):
patched_installers = []
for label, inst in installers:
def always_supports(self, ctx):
return True
def make_run(label_name: str):
def _run(self, ctx):
called_installers.append(label_name)
return _run
inst.supports = always_supports.__get__(inst, inst.__class__) # type: ignore[assignment]
inst.run = make_run(label).__get__(inst, inst.__class__) # type: ignore[assignment]
patched_installers.append(inst)
with patch.object(install_mod, "INSTALLERS", patched_installers), patch.object(
install_mod, "get_repo_identifier", return_value="dummy-repo"
), patch.object(
install_mod, "get_repo_dir", return_value=repo_dir
), patch.object(
install_mod, "verify_repository", return_value=(True, [], None, None)
), patch.object(
install_mod, "clone_repos"
with (
patch.object(install_mod, "INSTALLERS", patched_installers),
patch.object(install_mod, "get_repo_identifier", return_value="dummy-repo"),
patch.object(install_mod, "get_repo_dir", return_value=repo_dir),
patch.object(
install_mod, "verify_repository", return_value=(True, [], None, None)
),
patch.object(install_mod, "clone_repos"),
):
install_repos(
selected_repos=selected_repos,

View File

@@ -29,10 +29,12 @@ class TestIntegrationReleasePublishHook(unittest.TestCase):
# Go through real parser to ensure CLI surface is wired correctly
args = self._parse(["release", "patch"])
with patch("pkgmgr.cli.commands.release.run_release") as m_release, patch(
"pkgmgr.cli.commands.release.run_publish"
) as m_publish, patch(
"pkgmgr.cli.commands.release.sys.stdin.isatty", return_value=False
with (
patch("pkgmgr.cli.commands.release.run_release") as m_release,
patch("pkgmgr.cli.commands.release.run_publish") as m_publish,
patch(
"pkgmgr.cli.commands.release.sys.stdin.isatty", return_value=False
),
):
handle_release(args=args, ctx=self._ctx(), selected=selected)
@@ -53,9 +55,10 @@ class TestIntegrationReleasePublishHook(unittest.TestCase):
args = self._parse(["release", "patch", "--no-publish"])
with patch("pkgmgr.cli.commands.release.run_release") as m_release, patch(
"pkgmgr.cli.commands.release.run_publish"
) as m_publish:
with (
patch("pkgmgr.cli.commands.release.run_release") as m_release,
patch("pkgmgr.cli.commands.release.run_publish") as m_publish,
):
handle_release(args=args, ctx=self._ctx(), selected=selected)
m_release.assert_called_once()

View File

@@ -24,7 +24,10 @@ class TestIntegrationReposCreatePreview(unittest.TestCase):
repositories_base_dir="/tmp/Repositories",
binaries_dir="/tmp/bin",
all_repositories=[],
config_merged={"directories": {"repositories": "/tmp/Repositories"}, "repositories": []},
config_merged={
"directories": {"repositories": "/tmp/Repositories"},
"repositories": [],
},
user_config_path="/tmp/user.yml",
)

View File

@@ -18,9 +18,13 @@ def _find_repo_root() -> Path:
"""
here = Path(__file__).resolve()
for parent in here.parents:
if (parent / "pyproject.toml").is_file() and (parent / "src" / "pkgmgr").is_dir():
if (parent / "pyproject.toml").is_file() and (
parent / "src" / "pkgmgr"
).is_dir():
return parent
raise RuntimeError("Could not determine repository root for pkgmgr integration test")
raise RuntimeError(
"Could not determine repository root for pkgmgr integration test"
)
class TestRepositoryPathsExist(unittest.TestCase):

View File

@@ -25,7 +25,6 @@ class TestTokenResolverIntegration(unittest.TestCase):
# 1) ENV: empty
# ------------------------------------------------------------------
with patch.dict("os.environ", {}, clear=True):
# ------------------------------------------------------------------
# 2) GH CLI is available
# ------------------------------------------------------------------
@@ -37,14 +36,12 @@ class TestTokenResolverIntegration(unittest.TestCase):
"pkgmgr.core.credentials.providers.gh.subprocess.check_output",
return_value="gh-invalid-token\n",
):
# ------------------------------------------------------------------
# 3) Keyring returns an existing (invalid) token
# ------------------------------------------------------------------
with patch(
"pkgmgr.core.credentials.providers.keyring._import_keyring"
) as mock_import_keyring:
mock_keyring = mock_import_keyring.return_value
mock_keyring.get_password.return_value = "keyring-invalid-token"
@@ -59,7 +56,6 @@ class TestTokenResolverIntegration(unittest.TestCase):
"pkgmgr.core.credentials.providers.prompt.getpass",
return_value="new-valid-token",
):
# ------------------------------------------------------------------
# 5) Validation logic:
# - gh token invalid
@@ -77,7 +73,6 @@ class TestTokenResolverIntegration(unittest.TestCase):
"pkgmgr.core.credentials.resolver.validate_token",
side_effect=validate_side_effect,
) as validate_mock:
result = resolver.get_token(
provider_kind="github",
host="github.com",

View File

@@ -46,14 +46,22 @@ class TestUpdateSilentContinues(unittest.TestCase):
def install_side_effect(selected_repos, *_args, **kwargs):
repo = selected_repos[0]
install_calls.append((repo["repository"], kwargs.get("silent"), kwargs.get("emit_summary")))
install_calls.append(
(repo["repository"], kwargs.get("silent"), kwargs.get("emit_summary"))
)
if repo["repository"] == "repo-b":
raise SystemExit(3)
return None
# Patch at the exact import locations used inside UpdateManager.run()
with patch("pkgmgr.actions.repository.pull.pull_with_verification", side_effect=pull_side_effect), patch(
"pkgmgr.actions.install.install_repos", side_effect=install_side_effect
with (
patch(
"pkgmgr.actions.repository.pull.pull_with_verification",
side_effect=pull_side_effect,
),
patch(
"pkgmgr.actions.install.install_repos", side_effect=install_side_effect
),
):
# 1) silent=True: should NOT raise (even though failures happened)
UpdateManager().run(
@@ -73,7 +81,9 @@ class TestUpdateSilentContinues(unittest.TestCase):
# Ensure it tried all pulls, and installs happened for B and C only.
self.assertEqual(pull_calls, ["repo-a", "repo-b", "repo-c"])
self.assertEqual([r for r, _silent, _emit in install_calls], ["repo-b", "repo-c"])
self.assertEqual(
[r for r, _silent, _emit in install_calls], ["repo-b", "repo-c"]
)
# Ensure UpdateManager suppressed install summary spam by passing emit_summary=False.
for _repo_name, _silent, emit_summary in install_calls:
@@ -103,7 +113,9 @@ class TestUpdateSilentContinues(unittest.TestCase):
# Still must have processed all repos (continue-on-failure behavior).
self.assertEqual(pull_calls, ["repo-a", "repo-b", "repo-c"])
self.assertEqual([r for r, _silent, _emit in install_calls], ["repo-b", "repo-c"])
self.assertEqual(
[r for r, _silent, _emit in install_calls], ["repo-b", "repo-c"]
)
if __name__ == "__main__":