Fix path and shell repo directory resolution + add unit/E2E tests
- Introduce `_resolve_repository_directory()` to unify directory lookup
(explicit `directory` key → fallback to `get_repo_dir()` using base dir)
- Fix `pkgmgr path` to avoid KeyError and behave consistently with
other commands using lazy directory resolution
- Fix `pkgmgr shell` to use resolved directory and correctly emit cwd
- Add full E2E tests for `pkgmgr path --all` and `pkgmgr path pkgmgr`
- Add unit tests covering:
* explicit directory usage
* fallback resolution via get_repo_dir()
* empty selection behavior
* shell command cwd resolution
* missing shell command error handling
This commit is contained in:
@@ -16,10 +16,36 @@ from pkgmgr.actions.repository.list import list_repositories
|
||||
from pkgmgr.core.command.run import run_command
|
||||
from pkgmgr.actions.repository.create import create_repo
|
||||
from pkgmgr.core.repository.selected import get_selected_repos
|
||||
from pkgmgr.core.repository.dir import get_repo_dir
|
||||
|
||||
Repository = Dict[str, Any]
|
||||
|
||||
|
||||
def _resolve_repository_directory(repository: Repository, ctx: CLIContext) -> str:
|
||||
"""
|
||||
Resolve the local filesystem directory for a repository.
|
||||
|
||||
Priority:
|
||||
1. Use repository["directory"] if present.
|
||||
2. Fallback to get_repo_dir(...) using the repositories base directory
|
||||
from the CLI context.
|
||||
"""
|
||||
repo_dir = repository.get("directory")
|
||||
if repo_dir:
|
||||
return repo_dir
|
||||
|
||||
base_dir = (
|
||||
getattr(ctx, "repositories_base_dir", None)
|
||||
or getattr(ctx, "repositories_dir", None)
|
||||
)
|
||||
if not base_dir:
|
||||
raise RuntimeError(
|
||||
"Cannot resolve repositories base directory from context; "
|
||||
"expected ctx.repositories_base_dir or ctx.repositories_dir."
|
||||
)
|
||||
return get_repo_dir(base_dir, repository)
|
||||
|
||||
|
||||
def handle_repos_command(
|
||||
args,
|
||||
ctx: CLIContext,
|
||||
@@ -108,8 +134,25 @@ def handle_repos_command(
|
||||
# path
|
||||
# ------------------------------------------------------------
|
||||
if args.command == "path":
|
||||
if not selected:
|
||||
print("[pkgmgr] No repositories selected for path.")
|
||||
return
|
||||
|
||||
for repository in selected:
|
||||
print(repository["directory"])
|
||||
try:
|
||||
repo_dir = _resolve_repository_directory(repository, ctx)
|
||||
except Exception as exc:
|
||||
ident = (
|
||||
f"{repository.get('provider', '?')}/"
|
||||
f"{repository.get('account', '?')}/"
|
||||
f"{repository.get('repository', '?')}"
|
||||
)
|
||||
print(
|
||||
f"[WARN] Could not resolve directory for {ident}: {exc}"
|
||||
)
|
||||
continue
|
||||
|
||||
print(repo_dir)
|
||||
return
|
||||
|
||||
# ------------------------------------------------------------
|
||||
@@ -119,14 +162,14 @@ def handle_repos_command(
|
||||
if not args.shell_command:
|
||||
print("[ERROR] 'shell' requires a command via -c/--command.")
|
||||
sys.exit(2)
|
||||
|
||||
command_to_run = " ".join(args.shell_command)
|
||||
for repository in selected:
|
||||
print(
|
||||
f"Executing in '{repository['directory']}': {command_to_run}"
|
||||
)
|
||||
repo_dir = _resolve_repository_directory(repository, ctx)
|
||||
print(f"Executing in '{repo_dir}': {command_to_run}")
|
||||
run_command(
|
||||
command_to_run,
|
||||
cwd=repository["directory"],
|
||||
cwd=repo_dir,
|
||||
preview=args.preview,
|
||||
)
|
||||
return
|
||||
|
||||
117
tests/e2e/test_path_commands.py
Normal file
117
tests/e2e/test_path_commands.py
Normal file
@@ -0,0 +1,117 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
"""
|
||||
End-to-end tests for the `pkgmgr path` command.
|
||||
|
||||
We verify two usage patterns:
|
||||
|
||||
1) pkgmgr path --all
|
||||
- Should print the paths of all configured repositories.
|
||||
|
||||
2) pkgmgr path pkgmgr
|
||||
- Should print the path for the repository identified as "pkgmgr".
|
||||
|
||||
Both tests are considered successful if the command completes without
|
||||
raising an exception and exits with code 0 (or no explicit exit code).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import runpy
|
||||
import sys
|
||||
import unittest
|
||||
from contextlib import redirect_stdout
|
||||
|
||||
|
||||
class TestPathCommandsE2E(unittest.TestCase):
|
||||
def _run_pkgmgr_path(self, argv_tail: list[str]) -> str:
|
||||
"""
|
||||
Helper to run `pkgmgr path ...` via main.py and return stdout.
|
||||
|
||||
Args:
|
||||
argv_tail: List of arguments that follow the "pkgmgr" executable,
|
||||
e.g. ["path", "--all"] or ["path", "pkgmgr"].
|
||||
|
||||
Returns:
|
||||
The captured stdout produced by the command.
|
||||
|
||||
Raises:
|
||||
AssertionError if the command exits with a non-zero exit code.
|
||||
"""
|
||||
original_argv = sys.argv
|
||||
cmd_repr = "pkgmgr " + " ".join(argv_tail)
|
||||
buffer = io.StringIO()
|
||||
|
||||
try:
|
||||
sys.argv = ["pkgmgr"] + argv_tail
|
||||
|
||||
try:
|
||||
# Capture stdout while running the CLI entry point.
|
||||
with redirect_stdout(buffer):
|
||||
runpy.run_module("main", run_name="__main__")
|
||||
except SystemExit as exc:
|
||||
# Determine the exit code (int or string)
|
||||
exit_code = exc.code
|
||||
if isinstance(exit_code, int):
|
||||
numeric_code = exit_code
|
||||
else:
|
||||
try:
|
||||
numeric_code = int(exit_code)
|
||||
except (TypeError, ValueError):
|
||||
numeric_code = None
|
||||
|
||||
# Treat SystemExit(0) as success.
|
||||
if numeric_code == 0 or numeric_code is None:
|
||||
return buffer.getvalue()
|
||||
|
||||
# Non-zero exit code → fail with helpful message.
|
||||
raise AssertionError(
|
||||
f"{cmd_repr!r} failed with exit code {exit_code!r}. "
|
||||
"Scroll up to see the full pkgmgr output inside the container."
|
||||
) from exc
|
||||
|
||||
finally:
|
||||
sys.argv = original_argv
|
||||
|
||||
# No SystemExit raised → also treat as success.
|
||||
return buffer.getvalue()
|
||||
|
||||
def test_path_all_repositories(self) -> None:
|
||||
"""
|
||||
Run: pkgmgr path --all
|
||||
|
||||
The test succeeds if the command exits successfully and prints
|
||||
at least one non-empty line.
|
||||
"""
|
||||
output = self._run_pkgmgr_path(["path", "--all"])
|
||||
lines = [line for line in output.splitlines() if line.strip()]
|
||||
|
||||
# We only assert that something was printed; we do not assume
|
||||
# that repositories are already cloned on disk.
|
||||
self.assertGreater(
|
||||
len(lines),
|
||||
0,
|
||||
msg="Expected `pkgmgr path --all` to print at least one path.",
|
||||
)
|
||||
|
||||
def test_path_single_pkgmgr(self) -> None:
|
||||
"""
|
||||
Run: pkgmgr path pkgmgr
|
||||
|
||||
The test succeeds if the command exits successfully and prints
|
||||
at least one non-empty line (the resolved directory).
|
||||
"""
|
||||
output = self._run_pkgmgr_path(["path", "pkgmgr"])
|
||||
lines = [line for line in output.splitlines() if line.strip()]
|
||||
|
||||
self.assertGreater(
|
||||
len(lines),
|
||||
0,
|
||||
msg="Expected `pkgmgr path pkgmgr` to print at least one path.",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
216
tests/unit/pkgmgr/cli/commands/test_repos.py
Normal file
216
tests/unit/pkgmgr/cli/commands/test_repos.py
Normal file
@@ -0,0 +1,216 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
"""
|
||||
Unit tests for pkgmgr.cli.commands.repos
|
||||
|
||||
We focus on the behaviour of:
|
||||
|
||||
- _resolve_repository_directory(...)
|
||||
- handle_repos_command(...) for the "path" and "shell" commands
|
||||
|
||||
Goals:
|
||||
|
||||
* "path" should:
|
||||
- print repo["directory"] if present
|
||||
- fall back to get_repo_dir(ctx.repositories_base_dir, repo) otherwise
|
||||
- handle "no selected repos" gracefully
|
||||
|
||||
* "shell" should:
|
||||
- resolve the directory via _resolve_repository_directory(...)
|
||||
- call run_command(...) with cwd set to the resolved directory
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import sys
|
||||
import unittest
|
||||
from contextlib import redirect_stdout
|
||||
from types import SimpleNamespace
|
||||
from typing import Any, Dict, List
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from pkgmgr.cli.context import CLIContext
|
||||
from pkgmgr.cli.commands.repos import handle_repos_command
|
||||
|
||||
|
||||
Repository = Dict[str, Any]
|
||||
|
||||
|
||||
class TestReposCommand(unittest.TestCase):
|
||||
def _make_ctx(self, repositories: List[Repository]) -> CLIContext:
|
||||
"""
|
||||
Helper to build a minimal CLIContext for tests.
|
||||
"""
|
||||
return CLIContext(
|
||||
config_merged={},
|
||||
repositories_base_dir="/base/dir",
|
||||
all_repositories=repositories,
|
||||
binaries_dir="/bin/dir",
|
||||
user_config_path="~/.config/pkgmgr/config.yaml",
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# "path" command tests
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_path_uses_explicit_directory_if_present(self) -> None:
|
||||
"""
|
||||
When repository["directory"] is present, handle_repos_command("path")
|
||||
should print this value directly without calling get_repo_dir().
|
||||
"""
|
||||
repos: List[Repository] = [
|
||||
{
|
||||
"provider": "github.com",
|
||||
"account": "kevinveenbirkenbach",
|
||||
"repository": "package-manager",
|
||||
"directory": "/custom/path/pkgmgr",
|
||||
}
|
||||
]
|
||||
ctx = self._make_ctx(repos)
|
||||
|
||||
args = SimpleNamespace(
|
||||
command="path",
|
||||
preview=False,
|
||||
list=False,
|
||||
system=False,
|
||||
extra_args=[],
|
||||
)
|
||||
|
||||
buf = io.StringIO()
|
||||
|
||||
with patch(
|
||||
"pkgmgr.cli.commands.repos.get_repo_dir"
|
||||
) as mock_get_repo_dir, redirect_stdout(buf):
|
||||
handle_repos_command(args, ctx, selected=repos)
|
||||
|
||||
output = buf.getvalue().strip().splitlines()
|
||||
self.assertIn("/custom/path/pkgmgr", output)
|
||||
mock_get_repo_dir.assert_not_called()
|
||||
|
||||
def test_path_falls_back_to_get_repo_dir_if_directory_missing(self) -> None:
|
||||
"""
|
||||
When repository["directory"] is missing, handle_repos_command("path")
|
||||
should call get_repo_dir(ctx.repositories_base_dir, repo) and print
|
||||
the returned value.
|
||||
"""
|
||||
repos: List[Repository] = [
|
||||
{
|
||||
"provider": "github.com",
|
||||
"account": "kevinveenbirkenbach",
|
||||
"repository": "package-manager",
|
||||
}
|
||||
]
|
||||
ctx = self._make_ctx(repos)
|
||||
|
||||
args = SimpleNamespace(
|
||||
command="path",
|
||||
preview=False,
|
||||
list=False,
|
||||
system=False,
|
||||
extra_args=[],
|
||||
)
|
||||
|
||||
buf = io.StringIO()
|
||||
|
||||
with patch(
|
||||
"pkgmgr.cli.commands.repos.get_repo_dir",
|
||||
return_value="/resolved/from/get_repo_dir",
|
||||
) as mock_get_repo_dir, redirect_stdout(buf):
|
||||
handle_repos_command(args, ctx, selected=repos)
|
||||
|
||||
output = buf.getvalue().strip().splitlines()
|
||||
self.assertIn("/resolved/from/get_repo_dir", output)
|
||||
mock_get_repo_dir.assert_called_once_with("/base/dir", repos[0])
|
||||
|
||||
def test_path_with_no_selected_repos_prints_message(self) -> None:
|
||||
"""
|
||||
When 'selected' is empty, the 'path' command should print a friendly
|
||||
message and not raise.
|
||||
"""
|
||||
ctx = self._make_ctx(repositories=[])
|
||||
args = SimpleNamespace(
|
||||
command="path",
|
||||
preview=False,
|
||||
list=False,
|
||||
system=False,
|
||||
extra_args=[],
|
||||
)
|
||||
|
||||
buf = io.StringIO()
|
||||
with redirect_stdout(buf):
|
||||
handle_repos_command(args, ctx, selected=[])
|
||||
|
||||
output = buf.getvalue()
|
||||
self.assertIn("No repositories selected for path", output)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# "shell" command tests
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_shell_resolves_directory_and_calls_run_command(self) -> None:
|
||||
"""
|
||||
'shell' should resolve the repository directory and pass it as cwd
|
||||
to run_command(), along with the full shell command string.
|
||||
"""
|
||||
repos: List[Repository] = [
|
||||
{
|
||||
"provider": "github.com",
|
||||
"account": "kevinveenbirkenbach",
|
||||
"repository": "package-manager",
|
||||
}
|
||||
]
|
||||
ctx = self._make_ctx(repos)
|
||||
|
||||
args = SimpleNamespace(
|
||||
command="shell",
|
||||
preview=False,
|
||||
shell_command=["echo", "hello"],
|
||||
)
|
||||
|
||||
with patch(
|
||||
"pkgmgr.cli.commands.repos.get_repo_dir",
|
||||
return_value="/resolved/for/shell",
|
||||
) as mock_get_repo_dir, patch(
|
||||
"pkgmgr.cli.commands.repos.run_command"
|
||||
) as mock_run_command:
|
||||
buf = io.StringIO()
|
||||
with redirect_stdout(buf):
|
||||
handle_repos_command(args, ctx, selected=repos)
|
||||
|
||||
# _resolve_repository_directory should have called get_repo_dir
|
||||
mock_get_repo_dir.assert_called_once_with("/base/dir", repos[0])
|
||||
|
||||
# run_command should be invoked with cwd set to the resolved path
|
||||
mock_run_command.assert_called_once()
|
||||
called_args, called_kwargs = mock_run_command.call_args
|
||||
|
||||
self.assertEqual("echo hello", called_args[0]) # command string
|
||||
self.assertEqual("/resolved/for/shell", called_kwargs["cwd"])
|
||||
self.assertFalse(called_kwargs["preview"])
|
||||
|
||||
def test_shell_without_command_exits_with_error(self) -> None:
|
||||
"""
|
||||
'shell' without -c/--command should print an error and exit with code 2.
|
||||
"""
|
||||
repos: List[Repository] = []
|
||||
ctx = self._make_ctx(repos)
|
||||
|
||||
args = SimpleNamespace(
|
||||
command="shell",
|
||||
preview=False,
|
||||
shell_command=[],
|
||||
)
|
||||
|
||||
buf = io.StringIO()
|
||||
with redirect_stdout(buf), self.assertRaises(SystemExit) as cm:
|
||||
handle_repos_command(args, ctx, selected=repos)
|
||||
|
||||
self.assertEqual(cm.exception.code, 2)
|
||||
output = buf.getvalue()
|
||||
self.assertIn("'shell' requires a command via -c/--command", output)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user