rob/CmdForge
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(gui): reconcile provider install dialog with agent-harness-setup helper

Browse Source
This commit is contained in:
rob
2026-09-11 22:55:42 -03:00
parent 60cd44bcfb
commit 4bd0a7cc02
2 changed files with 637 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cmdforge/gui/dialogs/provider_install_dialog.py
+223 -75
View File
@@ -1,26 +1,37 @@
"""Provider installation wizard dialog."""
"""Provider installation wizard dialog.
Reconciled in Packet S6 to delegate harness installation to agent-harness-setup
via a dedicated terminal window while preserving the GUI configuration and
provider variants management workflows.
"""
import json
import os
import shutil
import subprocess
import tempfile
import uuid
from pathlib import Path
from typing import Optional
from PySide6.QtWidgets import (
QDialog, QVBoxLayout, QHBoxLayout, QLabel, QPushButton,
QListWidget, QListWidgetItem, QTextEdit, QProgressBar,
QMessageBox, QGroupBox, QWidget, QStackedWidget
)
from PySide6.QtCore import Qt, QThread, Signal, QProcess
from PySide6.QtCore import Qt, QThread, Signal
from PySide6.QtGui import QFont
from ...providers import load_providers, add_provider, Provider
from ...cli.provider_commands import _find_setup_helper
# Provider installation info (shared with CLI)
PROVIDER_INSTALL_INFO = {
# Supported AI provider metadata and variants (installation catalogue retired and
# delegated to agent-harness-setup)
SUPPORTED_PROVIDERS = {
"claude": {
"harness_id": "claude-code",
"group": "Anthropic Claude",
"install_cmd": "curl -fsSL https://claude.ai/install.sh | bash",
"requires": "curl and bash",
"setup": "Run 'claude' to sign in with a subscription or API key.",
"cost": "Paid subscription or API key",
"variants": [
@@ -31,29 +42,27 @@ PROVIDER_INSTALL_INFO = {
],
},
"codex": {
"harness_id": "codex",
"group": "OpenAI Codex",
"install_cmd": "npm install -g @openai/codex",
"requires": "Node.js 18+ and npm",
"setup": "Run 'codex' to sign in.",
"setup": "Run 'codex' to sign in. ChatGPT Plus/Pro includes Codex access.",
"cost": "Free tier or ChatGPT subscription, subject to account limits",
"variants": [
("codex", "codex exec -", "OpenAI Codex"),
],
},
"agy": {
"harness_id": "agy",
"group": "Google Antigravity",
"install_cmd": None,
"requires": "Google Antigravity desktop application",
"setup": "Install from https://antigravity.google/download, then run 'agy install' to configure your PATH.",
"cost": "Free tier available, subject to account limits",
"manual_only": True,
"variants": [
("agy", "agy -p", "Google Antigravity (automatic model selection)"),
],
},
"opencode": {
"harness_id": "opencode",
"group": "OpenCode",
"install_cmd": "curl -fsSL https://opencode.ai/install | bash",
"requires": "curl, bash",
"setup": "Run 'opencode' and use /connect to configure providers.",
"cost": "Free models available; paid providers require credentials",
"variants": [
@@ -64,9 +73,8 @@ PROVIDER_INSTALL_INFO = {
],
},
"crush": {
"harness_id": "crush",
"group": "Crush (Charm)",
"install_cmd": "npm install -g @charmland/crush",
"requires": "Node.js 18+ and npm",
"setup": "Run 'crush' to configure Hyper or your provider API keys.",
"cost": "Limited Hyper credits or your own provider billing",
"variants": [
@@ -74,9 +82,8 @@ PROVIDER_INSTALL_INFO = {
],
},
"ollama": {
"harness_id": "ollama",
"group": "Ollama (Local LLMs)",
"install_cmd": "curl -fsSL https://ollama.ai/install.sh | bash",
"requires": "curl, bash, 8GB+ RAM (GPU recommended)",
"setup": "Run 'ollama pull llama3.2' to download a model.",
"cost": "FREE (runs entirely on your machine)",
"variants": [
@@ -88,43 +95,123 @@ PROVIDER_INSTALL_INFO = {
},
}
# Retained alias for backwards compatibility
PROVIDER_INSTALL_INFO = SUPPORTED_PROVIDERS
class InstallWorker(QThread):
"""Background worker for running installation commands."""
def find_terminal_emulator() -> list[str] | None:
"""Find an available graphical terminal emulator command prefix."""
candidates = [
("x-terminal-emulator", ["-e"]),
("gnome-terminal", ["--"]),
("konsole", ["-e"]),
("xfce4-terminal", ["-e"]),
("kitty", []),
("alacritty", ["-e"]),
("xterm", ["-e"]),
]
for term, prefix in candidates:
bin_path = shutil.which(term)
if bin_path:
return [bin_path] + prefix
return None
class SetupTerminalWorker(QThread):
"""Background worker running agent-harness-setup in a dedicated terminal window."""
output = Signal(str)
finished = Signal(bool, str)
finished = Signal(bool, str, dict)
def __init__(self, command: str):
def __init__(
self,
helper_argv: list[str],
harness_id: str,
result_file: Path,
request_id: str,
terminal_launcher: list[str] | None = None,
):
super().__init__()
self.command = command
self.helper_argv = helper_argv
self.harness_id = harness_id
self.result_file = result_file
self.request_id = request_id
self.terminal_launcher = terminal_launcher
self._process: Optional[subprocess.Popen] = None
def run(self):
helper_cmd = list(self.helper_argv) + [
"--result-file", str(self.result_file),
"--request-id", self.request_id,
"--harness", self.harness_id,
]
if self.terminal_launcher:
full_cmd = list(self.terminal_launcher) + helper_cmd
else:
full_cmd = helper_cmd
try:
process = subprocess.Popen(
self.command,
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
self._process = subprocess.Popen(
full_cmd,
stdout=subprocess.PIPE if not self.terminal_launcher else None,
stderr=subprocess.STDOUT if not self.terminal_launcher else None,
text=True,
bufsize=1
)
for line in process.stdout:
if not self.terminal_launcher and self._process.stdout:
for line in self._process.stdout:
self.output.emit(line)
process.wait()
exit_code = self._process.wait()
if process.returncode == 0:
self.finished.emit(True, "Installation completed successfully!")
result_data = None
if self.result_file.is_file():
try:
result_data = json.loads(self.result_file.read_text(encoding="utf-8"))
except Exception:
result_data = None
# Validate request_id correlation
if result_data and result_data.get("request_id") != self.request_id:
self.finished.emit(False, "Security error: Request ID mismatch in setup result.", {})
return
status = result_data.get("status") if result_data else None
# Coherence validation (Contract §C.5)
if exit_code != 0 and status == "success":
self.finished.emit(False, f"Helper exited with error code {exit_code} despite reporting success.", result_data or {})
return
if exit_code == 0 and status in ("failed", "cancelled"):
self.finished.emit(False, "Setup was cancelled or reported failure.", result_data or {})
return
if exit_code == 0 and status == "success":
self.finished.emit(True, f"Setup for '{self.harness_id}' verified successfully!", result_data or {})
elif exit_code in (10, 130, 143) or status == "cancelled":
self.finished.emit(False, "Setup was cancelled by user.", result_data or {})
else:
self.finished.emit(False, f"Installation failed (exit code {process.returncode})")
diag = result_data.get("error_diagnostics") if result_data else None
msg = f"Setup failed: {diag}" if diag else f"Setup process exited with code {exit_code}."
self.finished.emit(False, msg, result_data or {})
except Exception as e:
self.finished.emit(False, f"Error: {e}")
self.finished.emit(False, f"Error launching setup helper: {e}", {})
def terminate_process(self):
"""Terminate the active setup process if running."""
if self._process and self._process.poll() is None:
try:
self._process.terminate()
except Exception:
pass
# Alias for backwards compatibility
InstallWorker = SetupTerminalWorker
class ProviderInstallDialog(QDialog):
"""Dialog for installing AI provider CLI tools."""
"""Dialog for installing AI provider CLI tools via agent-harness-setup."""
def __init__(self, parent=None):
super().__init__(parent)
@@ -132,8 +219,9 @@ class ProviderInstallDialog(QDialog):
self.setMinimumSize(700, 500)
self.resize(750, 550)
self._worker = None
self._selected_provider = None
self._worker: Optional[SetupTerminalWorker] = None
self._selected_provider: Optional[str] = None
self._temp_dir: Optional[str] = None
self._setup_ui()
self._load_providers()
@@ -172,7 +260,8 @@ class ProviderInstallDialog(QDialog):
layout.addWidget(title)
desc = QLabel(
"CmdForge works with various AI providers. Select one below to install it. "
"CmdForge works with various AI providers. Select one below to install and configure it. "
"Installation is performed interactively via the Agent Harness Setup helper. "
"Providers marked as [INSTALLED] are already available on your system."
)
desc.setWordWrap(True)
@@ -314,11 +403,8 @@ class ProviderInstallDialog(QDialog):
def _load_providers(self):
"""Load and display available providers."""
# Get currently configured providers
configured = {p.name for p in load_providers()}
# Check what's installed on the system
for key, info in PROVIDER_INSTALL_INFO.items():
self.provider_list.clear()
for key, info in SUPPORTED_PROVIDERS.items():
item = QListWidgetItem()
# Check if CLI tool is installed
@@ -337,9 +423,6 @@ class ProviderInstallDialog(QDialog):
def _check_installed(self, provider_key: str) -> bool:
"""Check if a provider's CLI tool is installed."""
info = PROVIDER_INSTALL_INFO[provider_key]
# Check common locations based on provider
if provider_key == "opencode":
opencode_path = Path.home() / ".opencode" / "bin" / "opencode"
return opencode_path.exists() or shutil.which("opencode") is not None
@@ -354,30 +437,29 @@ class ProviderInstallDialog(QDialog):
"""Handle provider selection."""
key = item.data(Qt.UserRole)
is_installed = item.data(Qt.UserRole + 1)
info = PROVIDER_INSTALL_INFO[key]
info = SUPPORTED_PROVIDERS[key]
self._selected_provider = key
self.btn_install.setEnabled(True)
# Update details
install_method = info['install_cmd'] or "Manual download"
is_manual = info.get("manual_only", False)
install_method = "Manual desktop setup" if is_manual else "Agent Harness Setup Helper (terminal)"
details = f"""<b>{info['group']}</b><br><br>
<b>Cost:</b> {info['cost']}<br>
<b>Requirements:</b> {info['requires']}<br>
<b>Install method:</b> <code>{install_method}</code><br>
<b>After install:</b> {info['setup']}<br><br>
<b>Available models:</b> {', '.join(v[0] for v in info['variants'])}"""
if is_installed:
details += "<br><br><span style='color: green;'>This provider is already installed. Click Install to reinstall or update.</span>"
details += "<br><br><span style='color: green;'>This provider is already installed. Click Reinstall to run setup again.</span>"
self.details_label.setText(details)
# Change button text based on installed status
if info["install_cmd"]:
self.btn_install.setText("Reinstall" if is_installed else "Install Selected")
else:
if is_manual:
self.btn_install.setText("Add Providers" if is_installed else "Setup Instructions")
else:
self.btn_install.setText("Reinstall" if is_installed else "Install Selected")
def _on_provider_double_clicked(self, item: QListWidgetItem):
"""Handle double-click to start install."""
@@ -389,9 +471,9 @@ class ProviderInstallDialog(QDialog):
if not self._selected_provider:
return
info = PROVIDER_INSTALL_INFO[self._selected_provider]
info = SUPPORTED_PROVIDERS[self._selected_provider]
if not info["install_cmd"]:
if info.get("manual_only"):
if self._check_installed(self._selected_provider):
self._populate_variants()
self.stack.setCurrentIndex(2)
@@ -399,30 +481,65 @@ class ProviderInstallDialog(QDialog):
QMessageBox.information(self, "Manual Setup", info["setup"])
return
helper_argv = _find_setup_helper()
if not helper_argv:
QMessageBox.warning(
self,
"Setup Helper Missing",
"Agent Harness Setup Helper ('agent-harness-setup') was not found on your PATH or in active environments.\n\n"
"Please install the helper or run manually:\n"
" pip install -e /home/rob/PycharmProjects/agent-harness-setup\n\n"
"Or run directly:\n"
" PYTHONPATH=/home/rob/PycharmProjects/agent-harness-setup/src python3 -m agent_harness_setup"
)
return
# Prepare secure temporary directory and result file
self._temp_dir = tempfile.mkdtemp(prefix="cmdforge-harness-gui-")
try:
os.chmod(self._temp_dir, 0o700)
except OSError:
pass
result_file = Path(self._temp_dir) / "result.json"
request_id = str(uuid.uuid4())
harness_id = info.get("harness_id", self._selected_provider)
# Switch to install page
self.stack.setCurrentIndex(1)
self.install_title.setText(f"Installing {info['group']}...")
self.install_cmd_label.setText(f"Command: {info['install_cmd']}")
self.install_cmd_label.setText(f"Helper: agent-harness-setup --harness {harness_id}")
self.output_text.clear()
self.progress.setRange(0, 0)
self.btn_back.setEnabled(False)
self.btn_next.setEnabled(False)
# Start installation
self._worker = InstallWorker(info['install_cmd'])
term_launcher = find_terminal_emulator()
if term_launcher:
self.output_text.append(f"Dedicated terminal window launched ({Path(term_launcher[0]).name}).")
self.output_text.append("Please complete the preflight and approval steps in that window.\n")
else:
self.output_text.append("No graphical terminal emulator detected; running in background process.\n")
self._worker = SetupTerminalWorker(
helper_argv=helper_argv,
harness_id=harness_id,
result_file=result_file,
request_id=request_id,
terminal_launcher=term_launcher,
)
self._worker.output.connect(self._on_install_output)
self._worker.finished.connect(self._on_install_finished)
self._worker.finished.connect(self._on_setup_finished)
self._worker.start()
def _on_install_output(self, text: str):
"""Handle installation output."""
self.output_text.append(text.rstrip())
# Auto-scroll to bottom
scrollbar = self.output_text.verticalScrollBar()
scrollbar.setValue(scrollbar.maximum())
def _on_install_finished(self, success: bool, message: str):
"""Handle installation completion."""
def _on_setup_finished(self, success: bool, message: str, result_data: dict):
"""Handle setup helper process completion."""
self.progress.setRange(0, 1)
self.progress.setValue(1 if success else 0)
self.btn_back.setEnabled(True)
@@ -430,18 +547,39 @@ class ProviderInstallDialog(QDialog):
if success:
self.install_title.setText("Installation Complete!")
self.output_text.append(f"\n{'='*50}\n{message}")
# Advisory PATH updates
path_updates = result_data.get("path_updates") or []
if path_updates:
cur_path = os.environ.get("PATH", "")
new_dirs = [
str(Path(os.path.expanduser(p)))
for p in path_updates
if Path(os.path.expanduser(p)).is_dir() and str(Path(os.path.expanduser(p))) not in cur_path
]
if new_dirs:
os.environ["PATH"] = ":".join(new_dirs) + ":" + cur_path
# Update installed status for provider item in list
for i in range(self.provider_list.count()):
item = self.provider_list.item(i)
if item.data(Qt.UserRole) == self._selected_provider:
item.setData(Qt.UserRole + 1, True)
info = SUPPORTED_PROVIDERS[self._selected_provider]
item.setText(f"{info['group']} [INSTALLED]")
item.setForeground(Qt.darkGreen)
break
self.btn_next.setEnabled(True)
# Populate variants page
self._populate_variants()
self.stack.setCurrentIndex(2)
else:
self.install_title.setText("Installation Failed")
self.output_text.append(f"\n{'='*50}\nERROR: {message}")
self.install_title.setText("Installation Incomplete")
self.output_text.append(f"\n{'='*50}\n{message}")
QMessageBox.warning(
self, "Installation Failed",
f"The installation did not complete successfully.\n\n{message}\n\n"
"You may need to install manually. Check the output for details."
self,
"Installation Incomplete",
f"{message}\n\nYou can inspect the logs or retry.",
)
def _populate_variants(self):
@@ -449,7 +587,7 @@ class ProviderInstallDialog(QDialog):
if not self._selected_provider:
return
info = PROVIDER_INSTALL_INFO[self._selected_provider]
info = SUPPORTED_PROVIDERS[self._selected_provider]
self.variants_list.clear()
# Get currently configured provider names
@@ -458,6 +596,7 @@ class ProviderInstallDialog(QDialog):
for name, command, description in info['variants']:
item = QListWidgetItem(f"{name} - {description}")
item.setData(Qt.UserRole, (name, command, description))
self.variants_list.addItem(item)
# Pre-select if not already configured
if name not in configured:
@@ -466,8 +605,6 @@ class ProviderInstallDialog(QDialog):
item.setForeground(Qt.gray)
item.setText(f"{name} - {description} [already configured]")
self.variants_list.addItem(item)
# Update next steps
self.next_steps_label.setText(
f"<b>Important:</b> Before using these providers, you need to authenticate:<br><br>"
@@ -497,3 +634,14 @@ class ProviderInstallDialog(QDialog):
)
self.accept()
def closeEvent(self, event):
"""Clean up worker and temporary resources on close."""
if self._worker and self._worker.isRunning():
self._worker.terminate_process()
self._worker.wait(2000)
if self._temp_dir and Path(self._temp_dir).is_dir():
shutil.rmtree(self._temp_dir, ignore_errors=True)
super().closeEvent(event)
tests/test_gui_provider_install.py
+413
View File
@@ -0,0 +1,413 @@
"""Acceptance tests for Packet S6: CmdForge GUI Provider Install Dialog reconciliation."""
import json
import os
import sys
import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
os.environ["QT_QPA_PLATFORM"] = "offscreen"
from PySide6.QtCore import Qt
from PySide6.QtWidgets import QApplication, QMessageBox
# Ensure single application instance for tests
_qapp = QApplication.instance()
if _qapp is None:
_qapp = QApplication(["cmdforge-gui-tests", "-platform", "offscreen"])
from cmdforge.gui.dialogs.provider_install_dialog import (
PROVIDER_INSTALL_INFO,
SUPPORTED_PROVIDERS,
ProviderInstallDialog,
SetupTerminalWorker,
find_terminal_emulator,
)
from cmdforge.providers import Provider, get_provider, load_providers, save_providers
@pytest.fixture
def temp_providers_file(tmp_path):
providers_file = tmp_path / ".cmdforge" / "providers.yaml"
with patch("cmdforge.providers.PROVIDERS_FILE", providers_file):
save_providers([])
yield providers_file
def test_supported_providers_catalogue_reconciliation():
"""Verify that duplicate raw shell install commands and requirements are retired."""
assert "claude" in SUPPORTED_PROVIDERS
assert "codex" in SUPPORTED_PROVIDERS
assert "agy" in SUPPORTED_PROVIDERS
assert "opencode" in SUPPORTED_PROVIDERS
assert "crush" in SUPPORTED_PROVIDERS
assert "ollama" in SUPPORTED_PROVIDERS
# Ensure no entry has raw install_cmd shell scripts or requires strings
for name, info in SUPPORTED_PROVIDERS.items():
assert "install_cmd" not in info, f"Provider {name} still contains raw install_cmd"
assert "requires" not in info, f"Provider {name} still contains requires"
assert "group" in info
assert "setup" in info
assert "cost" in info
assert "variants" in info
assert len(info["variants"]) > 0
# Ensure backwards compatibility alias
assert PROVIDER_INSTALL_INFO is SUPPORTED_PROVIDERS
def test_find_terminal_emulator():
"""Verify terminal emulator detection logic."""
with patch("shutil.which", side_effect=lambda cmd: "/usr/bin/gnome-terminal" if cmd == "gnome-terminal" else None):
launcher = find_terminal_emulator()
assert launcher == ["/usr/bin/gnome-terminal", "--"]
with patch("shutil.which", side_effect=lambda cmd: "/usr/bin/x-terminal-emulator" if cmd == "x-terminal-emulator" else None):
launcher = find_terminal_emulator()
assert launcher == ["/usr/bin/x-terminal-emulator", "-e"]
with patch("shutil.which", return_value=None):
launcher = find_terminal_emulator()
assert launcher is None
def test_dialog_init_and_provider_selection(temp_providers_file):
"""Verify dialog initialization and detail rendering for automated and manual providers."""
with patch("shutil.which", return_value=None):
dialog = ProviderInstallDialog()
assert dialog.provider_list.count() == len(SUPPORTED_PROVIDERS)
assert not dialog.btn_install.isEnabled()
# Select Claude (automated setup helper)
claude_item = None
for i in range(dialog.provider_list.count()):
item = dialog.provider_list.item(i)
if item.data(Qt.UserRole) == "claude":
claude_item = item
break
assert claude_item is not None
dialog._on_provider_selected(claude_item)
assert dialog._selected_provider == "claude"
assert dialog.btn_install.isEnabled()
assert dialog.btn_install.text() == "Install Selected"
assert "Agent Harness Setup Helper (terminal)" in dialog.details_label.text()
assert "Anthropic Claude" in dialog.details_label.text()
# Select AGY (manual setup instructions)
agy_item = None
for i in range(dialog.provider_list.count()):
item = dialog.provider_list.item(i)
if item.data(Qt.UserRole) == "agy":
agy_item = item
break
assert agy_item is not None
dialog._on_provider_selected(agy_item)
assert dialog._selected_provider == "agy"
assert dialog.btn_install.isEnabled()
assert dialog.btn_install.text() == "Setup Instructions"
assert "Manual desktop setup" in dialog.details_label.text()
dialog.close()
def test_start_install_missing_helper_shows_warning(temp_providers_file):
"""Verify missing setup helper displays warning and does not advance page."""
with patch("shutil.which", return_value=None):
dialog = ProviderInstallDialog()
dialog._selected_provider = "claude"
with (
patch("cmdforge.gui.dialogs.provider_install_dialog._find_setup_helper", return_value=None),
patch.object(QMessageBox, "warning") as mock_warn,
):
dialog._start_install()
assert mock_warn.called
assert "Agent Harness Setup Helper ('agent-harness-setup') was not found" in mock_warn.call_args[0][2]
# Verify dialog stayed on Page 0 (Selection)
assert dialog.stack.currentIndex() == 0
dialog.close()
def test_setup_terminal_worker_success(tmp_path):
"""Test SetupTerminalWorker success path with valid result correlation."""
result_file = tmp_path / "result.json"
req_id = "test-uuid-1234"
data = {
"status": "success",
"request_id": req_id,
"harness_id": "claude-code",
"path_updates": ["/custom/bin"],
}
class MockProcess:
def __init__(self, *args, **kwargs):
self.stdout = None
result_file.write_text(json.dumps(data), encoding="utf-8")
def wait(self):
return 0
def poll(self):
return 0
worker = SetupTerminalWorker(
helper_argv=["/bin/agent-harness-setup"],
harness_id="claude-code",
result_file=result_file,
request_id=req_id,
terminal_launcher=["/usr/bin/xterm", "-e"],
)
results = []
worker.finished.connect(lambda s, m, d: results.append((s, m, d)))
with patch("subprocess.Popen", side_effect=MockProcess):
worker.run()
assert len(results) == 1
success, msg, res_dict = results[0]
assert success is True
assert "verified successfully" in msg
assert res_dict["path_updates"] == ["/custom/bin"]
def test_setup_terminal_worker_security_mismatch(tmp_path):
"""Test SetupTerminalWorker rejects result with request_id mismatch."""
result_file = tmp_path / "result.json"
data = {
"status": "success",
"request_id": "wrong-uuid",
"harness_id": "claude-code",
}
class MockProcess:
def __init__(self, *args, **kwargs):
self.stdout = None
result_file.write_text(json.dumps(data), encoding="utf-8")
def wait(self):
return 0
worker = SetupTerminalWorker(
helper_argv=["/bin/agent-harness-setup"],
harness_id="claude-code",
result_file=result_file,
request_id="expected-uuid",
)
results = []
worker.finished.connect(lambda s, m, d: results.append((s, m, d)))
with patch("subprocess.Popen", side_effect=MockProcess):
worker.run()
assert len(results) == 1
success, msg, _ = results[0]
assert success is False
assert "Security error" in msg
def test_setup_terminal_worker_incoherent_exit(tmp_path):
"""Test SetupTerminalWorker rejects exit code 1 with status 'success'."""
result_file = tmp_path / "result.json"
req_id = "test-uuid"
data = {
"status": "success",
"request_id": req_id,
}
class MockProcess:
def __init__(self, *args, **kwargs):
self.stdout = None
result_file.write_text(json.dumps(data), encoding="utf-8")
def wait(self):
return 1
worker = SetupTerminalWorker(
helper_argv=["/bin/agent-harness-setup"],
harness_id="claude-code",
result_file=result_file,
request_id=req_id,
)
results = []
worker.finished.connect(lambda s, m, d: results.append((s, m, d)))
with patch("subprocess.Popen", side_effect=MockProcess):
worker.run()
assert len(results) == 1
success, msg, _ = results[0]
assert success is False
assert "error code 1" in msg
def test_setup_terminal_worker_cancelled(tmp_path):
"""Test SetupTerminalWorker handles user cancellation cleanly."""
result_file = tmp_path / "result.json"
req_id = "cancel-uuid"
data = {
"status": "cancelled",
"request_id": req_id,
}
class MockProcess:
def __init__(self, *args, **kwargs):
self.stdout = None
result_file.write_text(json.dumps(data), encoding="utf-8")
def wait(self):
return 10
worker = SetupTerminalWorker(
helper_argv=["/bin/agent-harness-setup"],
harness_id="claude-code",
result_file=result_file,
request_id=req_id,
)
results = []
worker.finished.connect(lambda s, m, d: results.append((s, m, d)))
with patch("subprocess.Popen", side_effect=MockProcess):
worker.run()
assert len(results) == 1
success, msg, _ = results[0]
assert success is False
assert "cancelled" in msg.lower()
def test_setup_terminal_worker_failure_with_diagnostics(tmp_path):
"""Test SetupTerminalWorker handles failure with error diagnostics."""
result_file = tmp_path / "result.json"
req_id = "fail-uuid"
data = {
"status": "failed",
"request_id": req_id,
"error_diagnostics": "Missing npm dependency",
}
class MockProcess:
def __init__(self, *args, **kwargs):
self.stdout = None
result_file.write_text(json.dumps(data), encoding="utf-8")
def wait(self):
return 1
worker = SetupTerminalWorker(
helper_argv=["/bin/agent-harness-setup"],
harness_id="claude-code",
result_file=result_file,
request_id=req_id,
)
results = []
worker.finished.connect(lambda s, m, d: results.append((s, m, d)))
with patch("subprocess.Popen", side_effect=MockProcess):
worker.run()
assert len(results) == 1
success, msg, _ = results[0]
assert success is False
assert "Missing npm dependency" in msg
def test_dialog_full_flow_success_to_variants_and_add(temp_providers_file, tmp_path):
"""Test full dialog flow: install succeeds, advances to variants, registers selected variants."""
with patch("shutil.which", return_value=None):
dialog = ProviderInstallDialog()
dialog._selected_provider = "claude"
mock_worker = MagicMock()
with (
patch("cmdforge.gui.dialogs.provider_install_dialog._find_setup_helper", return_value=["/bin/agent-harness-setup"]),
patch("cmdforge.gui.dialogs.provider_install_dialog.SetupTerminalWorker", return_value=mock_worker),
):
dialog._start_install()
assert dialog.stack.currentIndex() == 1 # Install page
assert "Installing Anthropic Claude" in dialog.install_title.text()
# Create dummy directory for path update test
custom_bin = tmp_path / "bin"
custom_bin.mkdir(parents=True, exist_ok=True)
result_data = {
"status": "success",
"request_id": "any",
"harness_id": "claude-code",
"path_updates": [str(custom_bin)],
}
dialog._on_setup_finished(True, "Setup verified!", result_data)
assert str(custom_bin) in os.environ.get("PATH", "")
assert dialog.stack.currentIndex() == 2
assert dialog.variants_list.count() == len(SUPPORTED_PROVIDERS["claude"]["variants"])
selected_items = [dialog.variants_list.item(i) for i in range(dialog.variants_list.count()) if dialog.variants_list.item(i).isSelected()]
assert len(selected_items) > 0
with patch.object(QMessageBox, "information") as mock_info:
dialog._add_selected_variants()
assert mock_info.called
provider = get_provider("claude")
assert provider is not None
assert provider.command == "claude -p"
dialog.close()
def test_dialog_failure_flow_shows_warning(temp_providers_file):
"""Test dialog failure flow warns user and remains on install page."""
with patch("shutil.which", return_value=None):
dialog = ProviderInstallDialog()
dialog._selected_provider = "claude"
mock_worker = MagicMock()
with (
patch("cmdforge.gui.dialogs.provider_install_dialog._find_setup_helper", return_value=["/bin/agent-harness-setup"]),
patch("cmdforge.gui.dialogs.provider_install_dialog.SetupTerminalWorker", return_value=mock_worker),
):
dialog._start_install()
with patch.object(QMessageBox, "warning") as mock_warn:
dialog._on_setup_finished(False, "Compilation error in installer", {})
assert mock_warn.called
assert "Compilation error in installer" in mock_warn.call_args[0][2]
assert dialog.stack.currentIndex() == 1
assert "Installation Incomplete" in dialog.install_title.text()
assert dialog.btn_back.isEnabled()
dialog.close()
def test_dialog_close_cleans_up_resources(temp_providers_file):
"""Test closing dialog terminates active worker and removes temporary directory."""
with patch("shutil.which", return_value=None):
dialog = ProviderInstallDialog()
temp_dir = tempfile.mkdtemp(prefix="test-cleanup-")
dialog._temp_dir = temp_dir
mock_worker = MagicMock()
mock_worker.isRunning.return_value = True
dialog._worker = mock_worker
dialog.close()
assert mock_worker.terminate_process.called
assert not Path(temp_dir).exists()