Refine command resolution and symlink creation (see ChatGPT conversation: https://chatgpt.com/share/6936be2d-952c-800f-a1cd-7ce5438014ff)

This commit is contained in:
Kevin Veen-Birkenbach
2025-12-08 13:02:05 +01:00
parent f641b95d81
commit 15f3c1bcba
9 changed files with 800 additions and 59 deletions

View File

@@ -1,62 +1,79 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os import os
from pkgmgr.get_repo_identifier import get_repo_identifier from pkgmgr.get_repo_identifier import get_repo_identifier
from pkgmgr.get_repo_dir import get_repo_dir from pkgmgr.get_repo_dir import get_repo_dir
def create_ink(repo, repositories_base_dir, bin_dir, all_repos, quiet=False, preview=False):
"""
Creates a symbolic link for the repository's command.
Instead of creating an executable wrapper, this function creates a symlink def create_ink(repo, repositories_base_dir, bin_dir, all_repos,
that points to the command file within the repository (e.g., main.sh or main.py). quiet=False, preview=False):
It also ensures that the command file has executable permissions.
""" """
Create a symlink for the repository's command.
IMPORTANT:
This function is intentionally kept *simple*. All decision logic for
choosing the command lives inside resolve_command_for_repo().
Behavior:
- If repo["command"] is defined → create a symlink to it.
- If repo["command"] is missing or None → do NOT create a link.
"""
repo_identifier = get_repo_identifier(repo, all_repos) repo_identifier = get_repo_identifier(repo, all_repos)
repo_dir = get_repo_dir(repositories_base_dir, repo) repo_dir = get_repo_dir(repositories_base_dir, repo)
command = repo.get("command") command = repo.get("command")
if not command: if not command:
# Automatically detect main.sh or main.py:
main_sh = os.path.join(repo_dir, "main.sh")
main_py = os.path.join(repo_dir, "main.py")
if os.path.exists(main_sh):
command = main_sh
elif os.path.exists(main_py):
command = main_py
else:
if not quiet: if not quiet:
print(f"No command defined and neither main.sh nor main.py found in {repo_dir}. Skipping link creation.") print(f"No command resolved for '{repo_identifier}'. Skipping link.")
return return
# Ensure the command file is executable. link_path = os.path.join(bin_dir, repo_identifier)
if not preview:
if preview:
print(f"[Preview] Would link {link_path}{command}")
return
# Mark local repo scripts as executable if needed
try: try:
if os.path.realpath(command).startswith(os.path.realpath(repo_dir)):
os.chmod(command, 0o755) os.chmod(command, 0o755)
except Exception as e: except Exception as e:
if not quiet: if not quiet:
print(f"Failed to set executable permissions for {command}: {e}") print(f"Failed to set permissions on '{command}': {e}")
link_path = os.path.join(bin_dir, repo_identifier) # Create bin directory
if preview:
print(f"[Preview] Would create symlink '{link_path}' pointing to '{command}'.")
else:
os.makedirs(bin_dir, exist_ok=True) os.makedirs(bin_dir, exist_ok=True)
# Remove existing
if os.path.exists(link_path) or os.path.islink(link_path): if os.path.exists(link_path) or os.path.islink(link_path):
os.remove(link_path) os.remove(link_path)
os.symlink(command, link_path)
if not quiet:
print(f"Symlink for {repo_identifier} created at {link_path}.")
# Create the link
os.symlink(command, link_path)
if not quiet:
print(f"Symlink created: {link_path}{command}")
# ------------------------------------------------------------
# Optional alias support (same as before)
# ------------------------------------------------------------
alias_name = repo.get("alias") alias_name = repo.get("alias")
if alias_name: if alias_name:
if alias_name == repo_identifier:
print(f"Skipped alias link creation. Alias '{alias_name}' and repository identifier '{repo_identifier}' are the same.")
else:
alias_link_path = os.path.join(bin_dir, alias_name) alias_link_path = os.path.join(bin_dir, alias_name)
if alias_name == repo_identifier:
if not quiet:
print(f"Alias '{alias_name}' equals identifier. Skipping alias creation.")
return
try: try:
if os.path.exists(alias_link_path) or os.path.islink(alias_link_path): if os.path.exists(alias_link_path) or os.path.islink(alias_link_path):
os.remove(alias_link_path) os.remove(alias_link_path)
os.symlink(link_path, alias_link_path) os.symlink(link_path, alias_link_path)
if not quiet: if not quiet:
print(f"Alias '{alias_name}' has been set to point to {repo_identifier}.") print(f"Alias '{alias_name}' created → {repo_identifier}")
except Exception as e: except Exception as e:
if not quiet: if not quiet:
print(f"Error creating alias '{alias_name}': {e}") print(f"Error creating alias '{alias_name}': {e}")

View File

@@ -8,7 +8,8 @@ This module orchestrates the installation of repositories by:
1. Ensuring the repository directory exists (cloning if necessary). 1. Ensuring the repository directory exists (cloning if necessary).
2. Verifying the repository according to the configured policies. 2. Verifying the repository according to the configured policies.
3. Creating executable links using create_ink(). 3. Creating executable links using create_ink(), after resolving the
appropriate command via resolve_command_for_repo().
4. Running a sequence of modular installer components that handle 4. Running a sequence of modular installer components that handle
specific technologies or manifests (PKGBUILD, Nix flakes, Python specific technologies or manifests (PKGBUILD, Nix flakes, Python
via pyproject.toml, Makefile, OS-specific package metadata). via pyproject.toml, Makefile, OS-specific package metadata).
@@ -25,8 +26,8 @@ from pkgmgr.get_repo_dir import get_repo_dir
from pkgmgr.create_ink import create_ink from pkgmgr.create_ink import create_ink
from pkgmgr.verify import verify_repository from pkgmgr.verify import verify_repository
from pkgmgr.clone_repos import clone_repos from pkgmgr.clone_repos import clone_repos
from pkgmgr.context import RepoContext from pkgmgr.context import RepoContext
from pkgmgr.resolve_command import resolve_command_for_repo
# Installer implementations # Installer implementations
from pkgmgr.installers.os_packages import ( from pkgmgr.installers.os_packages import (
@@ -204,7 +205,24 @@ def install_repos(
update_dependencies=update_dependencies, update_dependencies=update_dependencies,
) )
# Create the symlink using create_ink before running installers. # ------------------------------------------------------------
# Resolve the command for this repository before creating the link.
# If no command is resolved, no link will be created.
# ------------------------------------------------------------
resolved_command = resolve_command_for_repo(
repo=repo,
repo_identifier=identifier,
repo_dir=repo_dir,
)
if resolved_command:
repo["command"] = resolved_command
else:
repo.pop("command", None)
# ------------------------------------------------------------
# Create the symlink using create_ink (if a command is set).
# ------------------------------------------------------------
create_ink( create_ink(
repo, repo,
repositories_base_dir, repositories_base_dir,
@@ -239,7 +257,7 @@ def install_repos(
continue continue
# ------------------------------------------------------------ # ------------------------------------------------------------
# Debug + aussagekräftiger Fehler bei Installer-Fail # Debug output + clear error if an installer fails
# ------------------------------------------------------------ # ------------------------------------------------------------
if not quiet: if not quiet:
print( print(
@@ -268,10 +286,9 @@ def install_repos(
f" pkgmgr install {identifier} --clone-mode shallow --no-verification" f" pkgmgr install {identifier} --clone-mode shallow --no-verification"
) )
# Re-raise, damit CLI/Test sauber fehlschlägt, # Re-raise so that CLI/tests fail clearly,
# aber nun mit deutlich mehr Kontext. # but now with much more context.
raise raise
# Nur wenn der Installer erfolgreich war, Capabilities mergen # Only merge capabilities if the installer succeeded
provided_capabilities.update(caps) provided_capabilities.update(caps)

113
pkgmgr/resolve_command.py Normal file
View File

@@ -0,0 +1,113 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Command resolver for repositories.
This module determines the correct command to expose via symlink.
It implements the following priority:
1. Explicit command in repo config → command
2. System package manager binary (/usr/...) → NO LINK (respect OS)
3. Nix profile binary (~/.nix-profile/bin/<id>) → command
4. Python / non-system console script on PATH → command
5. Fallback: repository's main.sh or main.py → command
6. If nothing is available → raise error
The actual symlink creation is handled by create_ink(). This resolver
only decides *what* should be used as the entrypoint, or whether no
link should be created at all.
"""
import os
import shutil
from typing import Optional
def resolve_command_for_repo(repo, repo_identifier: str, repo_dir: str) -> Optional[str]:
"""
Determine the command for this repository.
Returns:
str → path to the command (a symlink should be created)
None → do NOT create a link (e.g. system package already provides it)
On total failure (no suitable command found at any layer), this function
raises SystemExit with a descriptive error message.
"""
# ------------------------------------------------------------
# 1. Explicit command defined by repository config
# ------------------------------------------------------------
explicit = repo.get("command")
if explicit:
return explicit
home = os.path.expanduser("~")
def is_executable(path: str) -> bool:
return os.path.exists(path) and os.access(path, os.X_OK)
# ------------------------------------------------------------
# 2. System package manager binary via PATH
#
# If the binary lives under /usr/, we treat it as a system-managed
# package (e.g. installed via pacman/apt/yum). In that case, pkgmgr
# does NOT create a link at all and defers entirely to the OS.
# ------------------------------------------------------------
path_candidate = shutil.which(repo_identifier)
system_binary: Optional[str] = None
non_system_binary: Optional[str] = None
if path_candidate:
if path_candidate.startswith("/usr/"):
system_binary = path_candidate
else:
non_system_binary = path_candidate
if system_binary:
# Respect system package manager: do not create a link.
if repo.get("debug", False):
print(
f"[pkgmgr] System binary for '{repo_identifier}' found at "
f"{system_binary}; no symlink will be created."
)
return None
# ------------------------------------------------------------
# 3. Nix profile binary (~/.nix-profile/bin/<identifier>)
# ------------------------------------------------------------
nix_candidate = os.path.join(home, ".nix-profile", "bin", repo_identifier)
if is_executable(nix_candidate):
return nix_candidate
# ------------------------------------------------------------
# 4. Python / non-system console script on PATH
#
# Here we reuse the non-system PATH candidate (e.g. from a venv or
# a user-local install like ~/.local/bin). This is treated as a
# valid command target.
# ------------------------------------------------------------
if non_system_binary and is_executable(non_system_binary):
return non_system_binary
# ------------------------------------------------------------
# 5. Fallback: main.sh / main.py inside the repository
# ------------------------------------------------------------
main_sh = os.path.join(repo_dir, "main.sh")
main_py = os.path.join(repo_dir, "main.py")
if is_executable(main_sh):
return main_sh
if is_executable(main_py) or os.path.exists(main_py):
return main_py
# ------------------------------------------------------------
# 6. Nothing found → treat as a hard error
# ------------------------------------------------------------
raise SystemExit(
f"No executable command could be resolved for repository '{repo_identifier}'. "
"No explicit 'command' configured, no system-managed binary under /usr/, "
"no Nix profile binary, no non-system console script on PATH, and no "
"main.sh/main.py found in the repository."
)

View File

@@ -0,0 +1,132 @@
# Integration Test: Command Resolution & Link Creation
**File:** `tests/integration/test_install_repos_integration.py`
This integration test validates the *end-to-end* behavior of the pkgmgr installation pipeline:
1. Repository selection
2. Verification
3. **Command resolution (`resolve_command_for_repo`)**
4. **Symlink creation (`create_ink`)**
5. Installer execution order and skipping rules
The test sets up **two repositories**:
| Repository | Environment Condition | Expected Behavior |
| ------------- | ------------------------------- | ------------------------------------ |
| `repo-system` | `/usr/bin/tool-system` exists | System binary → **NO symlink** |
| `repo-nix` | Nix profile contains `tool-nix` | Link → `~/.nix-profile/bin/tool-nix` |
This confirms that pkgmgr respects system package managers, prefers Nix over fallback logic, and creates or skips symlinks appropriately.
---
## 🔧 **Command Selection Flowchart**
The integration test verifies that pkgmgr follows this exact decision tree:
```mermaid
flowchart TD
A[Start: install_repos()] --> B(resolve_command_for_repo)
B --> C{Explicit command in repo config?}
C -- Yes --> C1[Return explicit command]
C -- No --> D
D --> E{System binary under /usr/?}
E -- Yes --> E1[Return None → NO symlink]
E -- No --> F
F --> G{Nix profile binary exists?}
G -- Yes --> G1[Return Nix binary]
G -- No --> H
H --> I{Python/non-system PATH binary?}
I -- Yes --> I1[Return PATH binary]
I -- No --> J
J --> K{main.sh/main.py in repo?}
K -- Yes --> K1[Return fallback script]
K -- No --> L[Error: No command found]
L --> X[Abort installation of this repo]
C1 --> M[create_ink → create symlink]
G1 --> M
I1 --> M
K1 --> M
E1 --> N[Skip symlink creation]
```
The integration test specifically checks branches:
* **System binary → skip**
* **Nix binary → create link**
---
## 📐 **Behavior Matrix (Simplified Priority Model)**
| Priority | Layer / Condition | Action | Link Created? |
| -------- | ------------------------------------- | ----------- | ------------- |
| 1 | Explicit `command` in repo config | Use it | ✅ Yes |
| 2 | System binary under `/usr/bin/...` | Respect OS | ❌ No |
| 3 | Nix profile binary exists | Use it | ✅ Yes |
| 4 | Non-system PATH binary | Use it | ✅ Yes |
| 5 | Repo fallback (`main.sh` / `main.py`) | Use it | ✅ Yes |
| 6 | None of the above | Raise error | ❌ No |
The integration test hits row **2** and **3**.
---
## 🧪 What This Integration Test Ensures
### ✔ Correct orchestration
`install_repos()` calls components in the correct sequence and respects outputs from each stage.
### ✔ Correct command resolution
The test asserts that:
* System binaries suppress symlink creation.
* Nix binaries produce symlinks even if PATH is empty.
### ✔ Correct linking behavior
For the Nix repo:
* A symlink is created under the `bin_dir`.
* The symlink points exactly to `~/.nix-profile/bin/<identifier>`.
### ✔ Isolation
No real system binaries or actual Nix installation are required—the test uses deterministic patches.
---
## 🧩 Additional Notes
* The integration test covers only the *positive Nix case* and the *system binary skip case*.
More tests can be added later for:
* Python binary resolution
* Fallback to `main.py`
* Error case when no command can be resolved
* The test intentionally uses a **temporary HOME directory** to simulate isolated Nix profiles.
---
## ✅ Summary
This integration test validates that:
* **pkgmgr does not overwrite or override system binaries**
* **Nix takes precedence over PATH-based tools**
* **The symlink layer works correctly**
* **The installer pipeline continues normally even when command resolution skips symlink creation**
The file provides a reliable foundation for confirming that command resolution, linking, and installation orchestration are functioning exactly as designed.

View File

@@ -0,0 +1,179 @@
# tests/integration/test_install_repos_integration.py
import os
import tempfile
import unittest
from unittest.mock import patch
import pkgmgr.install_repos as install_module
from pkgmgr.install_repos import install_repos
from pkgmgr.installers.base import BaseInstaller
class DummyInstaller(BaseInstaller):
"""
Minimal installer used to ensure that the installation pipeline runs
without executing any real external commands.
"""
layer = None
def supports(self, ctx):
return True
def run(self, ctx):
return
class TestInstallReposIntegration(unittest.TestCase):
@patch("pkgmgr.install_repos.verify_repository")
@patch("pkgmgr.install_repos.clone_repos")
@patch("pkgmgr.install_repos.get_repo_dir")
@patch("pkgmgr.install_repos.get_repo_identifier")
def test_system_binary_vs_nix_binary(
self,
mock_get_repo_identifier,
mock_get_repo_dir,
mock_clone_repos,
mock_verify_repository,
):
"""
Full integration test for high-level command resolution + symlink creation.
We do NOT re-test all low-level file-system details of
resolve_command_for_repo here (that is covered by unit tests).
Instead, we assert that:
- If resolve_command_for_repo(...) returns None:
→ install_repos() does NOT create a symlink.
- If resolve_command_for_repo(...) returns a path:
→ install_repos() creates exactly one symlink in bin_dir
that points to this path.
Concretely:
- repo-system:
resolve_command_for_repo(...) → None
→ no symlink in bin_dir for this repo.
- repo-nix:
resolve_command_for_repo(...) → "/nix/profile/bin/repo-nix"
→ exactly one symlink in bin_dir pointing to that path.
"""
# Repositories must have provider/account/repository so that get_repo_dir()
# does not crash when called from create_ink().
repo_system = {
"name": "repo-system",
"provider": "github.com",
"account": "dummy",
"repository": "repo-system",
}
repo_nix = {
"name": "repo-nix",
"provider": "github.com",
"account": "dummy",
"repository": "repo-nix",
}
selected_repos = [repo_system, repo_nix]
all_repos = selected_repos
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")
os.makedirs(repo_system_dir, exist_ok=True)
os.makedirs(repo_nix_dir, exist_ok=True)
# Identifiers and repo dirs used inside install_repos()
mock_get_repo_identifier.side_effect = ["repo-system", "repo-nix"]
mock_get_repo_dir.side_effect = [repo_system_dir, repo_nix_dir]
# Repository verification always succeeds
mock_verify_repository.return_value = (True, [], "commit", "key")
mock_clone_repos.return_value = None
# Pretend this is the "Nix binary" path for repo-nix
nix_tool_path = "/nix/profile/bin/repo-nix"
# Patch resolve_command_for_repo at the install_repos module level
with patch("pkgmgr.install_repos.resolve_command_for_repo") as mock_resolve, \
patch("pkgmgr.install_repos.os.path.exists") as mock_exists_install:
def fake_resolve_command(repo, repo_identifier: str, repo_dir: str):
"""
High-level behavior stub:
- For repo-system: act as if a system package owns the command
→ return None (no symlink).
- For repo-nix: act as if a Nix profile binary is the entrypoint
→ return nix_tool_path (symlink should be created).
"""
if repo_identifier == "repo-system":
return None
if repo_identifier == "repo-nix":
return nix_tool_path
return None
def fake_exists_install(path: str) -> bool:
"""
Make _ensure_repo_dir() believe that the repo directories
already exist so that it does not attempt cloning.
"""
if path in (repo_system_dir, repo_nix_dir):
return True
return False
mock_resolve.side_effect = fake_resolve_command
mock_exists_install.side_effect = fake_exists_install
# Use only DummyInstaller so we focus on link creation, not installer behavior
old_installers = install_module.INSTALLERS
install_module.INSTALLERS = [DummyInstaller()]
try:
install_repos(
selected_repos=selected_repos,
repositories_base_dir=tmp_base,
bin_dir=tmp_bin,
all_repos=all_repos,
no_verification=False,
preview=False,
quiet=False,
clone_mode="shallow",
update_dependencies=False,
)
finally:
install_module.INSTALLERS = old_installers
# ------------------------------------------------------------------
# Inspect bin_dir: exactly one symlink must exist, pointing to Nix.
# ------------------------------------------------------------------
symlink_paths = []
for entry in os.listdir(tmp_bin):
full = os.path.join(tmp_bin, entry)
if os.path.islink(full):
symlink_paths.append(full)
# There must be exactly one symlink (for repo-nix)
self.assertEqual(
len(symlink_paths),
1,
f"Expected exactly one symlink in {tmp_bin}, found {symlink_paths}",
)
target = os.readlink(symlink_paths[0])
# That symlink must point to the "Nix" path returned by the resolver stub
self.assertEqual(
target,
nix_tool_path,
f"Expected symlink target to be Nix binary {nix_tool_path}, got {target}",
)
if __name__ == "__main__":
unittest.main()

View File

@@ -98,6 +98,11 @@ class TestRecursiveCapabilitiesIntegration(unittest.TestCase):
The installers' supports() are forced to True so that only the The installers' supports() are forced to True so that only the
capability-shadowing logic decides whether they are skipped. capability-shadowing logic decides whether they are skipped.
The installers' run() methods are patched to avoid real commands. The installers' run() methods are patched to avoid real commands.
NOTE:
We patch resolve_command_for_repo() to always return a dummy
command path so that command resolution does not interfere with
capability-layering tests.
""" """
if selected_repos is None: if selected_repos is None:
repo = {} repo = {}
@@ -128,7 +133,8 @@ class TestRecursiveCapabilitiesIntegration(unittest.TestCase):
patch.object(install_mod, "get_repo_dir", return_value=repo_dir), \ 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, "verify_repository", return_value=(True, [], None, None)), \
patch.object(install_mod, "create_ink"), \ patch.object(install_mod, "create_ink"), \
patch.object(install_mod, "clone_repos"): patch.object(install_mod, "clone_repos"), \
patch.object(install_mod, "resolve_command_for_repo", return_value="/bin/dummy"):
install_repos( install_repos(
selected_repos=selected_repos, selected_repos=selected_repos,
@@ -144,6 +150,7 @@ class TestRecursiveCapabilitiesIntegration(unittest.TestCase):
return called_installers return called_installers
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# Scenario 1: Only Makefile with install target # Scenario 1: Only Makefile with install target
# ------------------------------------------------------------------ # ------------------------------------------------------------------

View File

@@ -0,0 +1,102 @@
# tests/unit/pkgmgr/test_create_ink.py
import unittest
from unittest.mock import patch
import pkgmgr.create_ink as create_ink_module
class TestCreateInk(unittest.TestCase):
@patch("pkgmgr.create_ink.get_repo_dir")
@patch("pkgmgr.create_ink.get_repo_identifier")
def test_create_ink_skips_when_no_command(
self,
mock_get_repo_identifier,
mock_get_repo_dir,
):
repo = {} # no 'command' key
mock_get_repo_identifier.return_value = "test-id"
mock_get_repo_dir.return_value = "/repos/test-id"
with patch("pkgmgr.create_ink.os.makedirs") as mock_makedirs, \
patch("pkgmgr.create_ink.os.symlink") as mock_symlink, \
patch("pkgmgr.create_ink.os.chmod") as mock_chmod:
create_ink_module.create_ink(
repo=repo,
repositories_base_dir="/repos",
bin_dir="/bin",
all_repos=[repo],
quiet=True,
preview=False,
)
mock_makedirs.assert_not_called()
mock_symlink.assert_not_called()
mock_chmod.assert_not_called()
@patch("pkgmgr.create_ink.get_repo_dir")
@patch("pkgmgr.create_ink.get_repo_identifier")
def test_create_ink_preview_only(
self,
mock_get_repo_identifier,
mock_get_repo_dir,
):
repo = {"command": "/repos/test-id/main.py"}
mock_get_repo_identifier.return_value = "test-id"
mock_get_repo_dir.return_value = "/repos/test-id"
with patch("pkgmgr.create_ink.os.makedirs") as mock_makedirs, \
patch("pkgmgr.create_ink.os.symlink") as mock_symlink, \
patch("pkgmgr.create_ink.os.chmod") as mock_chmod:
create_ink_module.create_ink(
repo=repo,
repositories_base_dir="/repos",
bin_dir="/bin",
all_repos=[repo],
quiet=True,
preview=True,
)
mock_makedirs.assert_not_called()
mock_symlink.assert_not_called()
mock_chmod.assert_not_called()
@patch("pkgmgr.create_ink.get_repo_dir")
@patch("pkgmgr.create_ink.get_repo_identifier")
def test_create_ink_creates_symlink_and_alias(
self,
mock_get_repo_identifier,
mock_get_repo_dir,
):
repo = {
"command": "/repos/test-id/main.py",
"alias": "alias-id",
}
mock_get_repo_identifier.return_value = "test-id"
mock_get_repo_dir.return_value = "/repos/test-id"
with patch("pkgmgr.create_ink.os.makedirs") as mock_makedirs, \
patch("pkgmgr.create_ink.os.symlink") as mock_symlink, \
patch("pkgmgr.create_ink.os.chmod") as mock_chmod, \
patch("pkgmgr.create_ink.os.path.exists", return_value=False), \
patch("pkgmgr.create_ink.os.path.islink", return_value=False), \
patch("pkgmgr.create_ink.os.remove") as mock_remove, \
patch("pkgmgr.create_ink.os.path.realpath", side_effect=lambda p: p):
create_ink_module.create_ink(
repo=repo,
repositories_base_dir="/repos",
bin_dir="/bin",
all_repos=[repo],
quiet=True,
preview=False,
)
# main link + alias link
self.assertEqual(mock_symlink.call_count, 2)
mock_makedirs.assert_called_once()
mock_chmod.assert_called_once()
mock_remove.assert_not_called()
if __name__ == "__main__":
unittest.main()

View File

@@ -11,7 +11,7 @@ from pkgmgr.installers.base import BaseInstaller
class DummyInstaller(BaseInstaller): class DummyInstaller(BaseInstaller):
"""Simple installer for testing orchestration.""" """Simple installer for testing orchestration."""
layer = None # keine speziellen Capabilities layer = None # no specific capabilities
def __init__(self): def __init__(self):
self.calls = [] self.calls = []
@@ -26,6 +26,7 @@ class DummyInstaller(BaseInstaller):
class TestInstallReposOrchestration(unittest.TestCase): class TestInstallReposOrchestration(unittest.TestCase):
@patch("pkgmgr.install_repos.create_ink") @patch("pkgmgr.install_repos.create_ink")
@patch("pkgmgr.install_repos.resolve_command_for_repo")
@patch("pkgmgr.install_repos.verify_repository") @patch("pkgmgr.install_repos.verify_repository")
@patch("pkgmgr.install_repos.get_repo_dir") @patch("pkgmgr.install_repos.get_repo_dir")
@patch("pkgmgr.install_repos.get_repo_identifier") @patch("pkgmgr.install_repos.get_repo_identifier")
@@ -36,6 +37,7 @@ class TestInstallReposOrchestration(unittest.TestCase):
mock_get_repo_identifier, mock_get_repo_identifier,
mock_get_repo_dir, mock_get_repo_dir,
mock_verify_repository, mock_verify_repository,
mock_resolve_command_for_repo,
mock_create_ink, mock_create_ink,
): ):
repo1 = {"name": "repo1"} repo1 = {"name": "repo1"}
@@ -50,6 +52,9 @@ class TestInstallReposOrchestration(unittest.TestCase):
# Simulate verification success: (ok, errors, commit, key) # Simulate verification success: (ok, errors, commit, key)
mock_verify_repository.return_value = (True, [], "commit", "key") mock_verify_repository.return_value = (True, [], "commit", "key")
# Resolve commands for both repos so create_ink will be called
mock_resolve_command_for_repo.side_effect = ["/bin/cmd1", "/bin/cmd2"]
# Ensure directories exist (no cloning) # Ensure directories exist (no cloning)
with patch("os.path.exists", return_value=True): with patch("os.path.exists", return_value=True):
dummy_installer = DummyInstaller() dummy_installer = DummyInstaller()
@@ -75,6 +80,7 @@ class TestInstallReposOrchestration(unittest.TestCase):
self.assertEqual(dummy_installer.calls, ["id1", "id2"]) self.assertEqual(dummy_installer.calls, ["id1", "id2"])
self.assertEqual(mock_create_ink.call_count, 2) self.assertEqual(mock_create_ink.call_count, 2)
self.assertEqual(mock_verify_repository.call_count, 2) self.assertEqual(mock_verify_repository.call_count, 2)
self.assertEqual(mock_resolve_command_for_repo.call_count, 2)
@patch("pkgmgr.install_repos.verify_repository") @patch("pkgmgr.install_repos.verify_repository")
@patch("pkgmgr.install_repos.get_repo_dir") @patch("pkgmgr.install_repos.get_repo_dir")
@@ -100,6 +106,7 @@ class TestInstallReposOrchestration(unittest.TestCase):
dummy_installer = DummyInstaller() dummy_installer = DummyInstaller()
with patch("os.path.exists", return_value=True), \ with patch("os.path.exists", return_value=True), \
patch("pkgmgr.install_repos.create_ink") as mock_create_ink, \ patch("pkgmgr.install_repos.create_ink") as mock_create_ink, \
patch("pkgmgr.install_repos.resolve_command_for_repo") as mock_resolve_cmd, \
patch("builtins.input", return_value="n"): patch("builtins.input", return_value="n"):
old_installers = install_module.INSTALLERS old_installers = install_module.INSTALLERS
install_module.INSTALLERS = [dummy_installer] install_module.INSTALLERS = [dummy_installer]
@@ -121,6 +128,7 @@ class TestInstallReposOrchestration(unittest.TestCase):
# No installer run and no create_ink when user declines # No installer run and no create_ink when user declines
self.assertEqual(dummy_installer.calls, []) self.assertEqual(dummy_installer.calls, [])
mock_create_ink.assert_not_called() mock_create_ink.assert_not_called()
mock_resolve_cmd.assert_not_called()
if __name__ == "__main__": if __name__ == "__main__":

View File

@@ -0,0 +1,166 @@
# tests/unit/pkgmgr/test_resolve_command.py
import unittest
from unittest.mock import patch
import pkgmgr.resolve_command as resolve_command_module
class TestResolveCommandForRepo(unittest.TestCase):
def test_explicit_command_wins(self):
repo = {"command": "/custom/cmd"}
result = resolve_command_module.resolve_command_for_repo(
repo=repo,
repo_identifier="tool",
repo_dir="/repos/tool",
)
self.assertEqual(result, "/custom/cmd")
@patch("pkgmgr.resolve_command.shutil.which", return_value="/usr/bin/tool")
def test_system_binary_returns_none_and_no_error(self, mock_which):
repo = {}
result = resolve_command_module.resolve_command_for_repo(
repo=repo,
repo_identifier="tool",
repo_dir="/repos/tool",
)
# System binary → no link
self.assertIsNone(result)
@patch("pkgmgr.resolve_command.os.access")
@patch("pkgmgr.resolve_command.os.path.exists")
@patch("pkgmgr.resolve_command.shutil.which", return_value=None)
@patch("pkgmgr.resolve_command.os.path.expanduser", return_value="/fakehome")
def test_nix_profile_binary(
self,
mock_expanduser,
mock_which,
mock_exists,
mock_access,
):
"""
No system/PATH binary, but a Nix profile binary exists:
→ must return the Nix binary path.
"""
repo = {}
fake_home = "/fakehome"
nix_path = f"{fake_home}/.nix-profile/bin/tool"
def fake_exists(path):
# Only the Nix binary exists
return path == nix_path
def fake_access(path, mode):
# Only the Nix binary is executable
return path == nix_path
mock_exists.side_effect = fake_exists
mock_access.side_effect = fake_access
result = resolve_command_module.resolve_command_for_repo(
repo=repo,
repo_identifier="tool",
repo_dir="/repos/tool",
)
self.assertEqual(result, nix_path)
@patch("pkgmgr.resolve_command.os.access")
@patch("pkgmgr.resolve_command.os.path.exists")
@patch("pkgmgr.resolve_command.os.path.expanduser", return_value="/home/user")
@patch("pkgmgr.resolve_command.shutil.which", return_value="/home/user/.local/bin/tool")
def test_non_system_binary_on_path(
self,
mock_which,
mock_expanduser,
mock_exists,
mock_access,
):
"""
No system (/usr) binary and no Nix binary, but a non-system
PATH binary exists (e.g. venv or ~/.local/bin):
→ must return that PATH binary.
"""
repo = {}
non_system_path = "/home/user/.local/bin/tool"
nix_candidate = "/home/user/.nix-profile/bin/tool"
def fake_exists(path):
# Only the non-system PATH binary "exists".
return path == non_system_path
def fake_access(path, mode):
# Only the non-system PATH binary is executable.
return path == non_system_path
mock_exists.side_effect = fake_exists
mock_access.side_effect = fake_access
result = resolve_command_module.resolve_command_for_repo(
repo=repo,
repo_identifier="tool",
repo_dir="/repos/tool",
)
self.assertEqual(result, non_system_path)
@patch("pkgmgr.resolve_command.os.access")
@patch("pkgmgr.resolve_command.os.path.exists")
@patch("pkgmgr.resolve_command.shutil.which", return_value=None)
@patch("pkgmgr.resolve_command.os.path.expanduser", return_value="/fakehome")
def test_fallback_to_main_py(
self,
mock_expanduser,
mock_which,
mock_exists,
mock_access,
):
"""
No system/non-system PATH binary, no Nix binary, but main.py exists:
→ must fall back to main.py in the repo.
"""
repo = {}
main_py = "/repos/tool/main.py"
def fake_exists(path):
return path == main_py
def fake_access(path, mode):
return path == main_py
mock_exists.side_effect = fake_exists
mock_access.side_effect = fake_access
result = resolve_command_module.resolve_command_for_repo(
repo=repo,
repo_identifier="tool",
repo_dir="/repos/tool",
)
self.assertEqual(result, main_py)
@patch("pkgmgr.resolve_command.os.access", return_value=False)
@patch("pkgmgr.resolve_command.os.path.exists", return_value=False)
@patch("pkgmgr.resolve_command.shutil.which", return_value=None)
@patch("pkgmgr.resolve_command.os.path.expanduser", return_value="/fakehome")
def test_no_command_results_in_system_exit(
self,
mock_expanduser,
mock_which,
mock_exists,
mock_access,
):
"""
Nothing available at any layer:
→ must raise SystemExit with a descriptive error message.
"""
repo = {}
with self.assertRaises(SystemExit) as cm:
resolve_command_module.resolve_command_for_repo(
repo=repo,
repo_identifier="tool",
repo_dir="/repos/tool",
)
msg = str(cm.exception)
self.assertIn("No executable command could be resolved for repository 'tool'", msg)
if __name__ == "__main__":
unittest.main()