463 lines
15 KiB
Python
463 lines
15 KiB
Python
"""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
|
|
|
|
|
|
def _write_completion(path, data):
|
|
payload = {
|
|
"schema_version": 1, "helper_version": "0.1.0",
|
|
"timestamp": "2026-09-14T00:00:00Z", "harness_id": "claude-code",
|
|
"auth_status": "unknown", "path_updates": [], "manual_steps_remaining": [],
|
|
"partial_changes": {"observed_created": [], "observed_modified": [], "unknown_changes": True},
|
|
"installer_exit_code": 0,
|
|
"recipe_id": "test-recipe", "recipe_version": "2026-09-10", "install_method": "argv",
|
|
}
|
|
payload.update(data)
|
|
path.write_text(json.dumps(payload), encoding="utf-8")
|
|
path.chmod(0o600)
|
|
|
|
|
|
@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 = {
|
|
"schema_version": 1,
|
|
"helper_version": "0.1.0",
|
|
"timestamp": "2026-09-14T00:00:00Z",
|
|
"auth_status": "unknown",
|
|
"manual_steps_remaining": [],
|
|
"partial_changes": {"observed_created": [], "observed_modified": [], "unknown_changes": True},
|
|
"verification": {"verified": True, "executable_path": "/custom/bin/claude", "version_output": "test"},
|
|
"status": "success",
|
|
"request_id": req_id,
|
|
"harness_id": "claude-code",
|
|
"path_updates": ["/custom/bin"],
|
|
}
|
|
|
|
class MockProcess:
|
|
def __init__(self, *args, **kwargs):
|
|
self.stdout = None
|
|
_write_completion(result_file, data)
|
|
|
|
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"]
|
|
|
|
|
|
@pytest.mark.parametrize("overrides", [
|
|
{"schema_version": True}, {"unexpected": 1}, {"status": []},
|
|
{"partial_changes": None}, {"cleanup_complete": "true"},
|
|
])
|
|
def test_worker_rejects_malformed_contract(tmp_path, overrides):
|
|
path = tmp_path / "result.json"
|
|
data = {"status": "failed", "request_id": "req-1"}
|
|
data.update(overrides)
|
|
_write_completion(path, data)
|
|
process = MagicMock(stdout=None)
|
|
process.wait.return_value = 1
|
|
worker = SetupTerminalWorker(["fake-helper"], "claude-code", path, "req-1")
|
|
outcomes = []
|
|
worker.finished.connect(lambda *args: outcomes.append(args))
|
|
with patch("subprocess.Popen", return_value=process):
|
|
worker.run()
|
|
assert len(outcomes) == 1 and outcomes[0][0] is False
|
|
assert "unknown" in outcomes[0][1]
|
|
|
|
|
|
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
|
|
_write_completion(result_file, data)
|
|
|
|
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 "does not match the expected" 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
|
|
_write_completion(result_file, data)
|
|
|
|
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 "exit 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 = {
|
|
"schema_version": 1,
|
|
"helper_version": "0.1.0",
|
|
"timestamp": "2026-09-14T00:00:00Z",
|
|
"harness_id": "claude-code",
|
|
"auth_status": "unknown",
|
|
"path_updates": [],
|
|
"manual_steps_remaining": [],
|
|
"partial_changes": {"observed_created": [], "observed_modified": [], "unknown_changes": True},
|
|
"status": "cancelled",
|
|
"request_id": req_id,
|
|
}
|
|
|
|
class MockProcess:
|
|
def __init__(self, *args, **kwargs):
|
|
self.stdout = None
|
|
_write_completion(result_file, data)
|
|
|
|
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
|
|
_write_completion(result_file, data)
|
|
|
|
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()
|