201 lines
7.0 KiB
Python
201 lines
7.0 KiB
Python
import io
|
|
import signal
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from test_config import settings
|
|
|
|
from lai_automation.agent import AgentProviderError, AgentTimeoutError, AiderAgent
|
|
|
|
|
|
def test_aider_command_cannot_mutate_git_configuration_or_send_analytics() -> None:
|
|
command = AiderAgent(settings()).command(Path("/repo/.git/task.txt"))
|
|
|
|
assert "--no-gitignore" in command
|
|
assert "--no-analytics" in command
|
|
assert "--no-auto-commits" in command
|
|
assert "--no-git" in command
|
|
assert "--stream" in command
|
|
assert "--no-pretty" in command
|
|
assert "--no-check-update" in command
|
|
assert "--no-suggest-shell-commands" in command
|
|
assert "--no-detect-urls" in command
|
|
assert "--disable-playwright" in command
|
|
assert "--no-auto-lint" in command
|
|
assert command[command.index("--map-tokens") + 1] == "2048"
|
|
assert command.count("/dev/null") == 2
|
|
assert "--edit-format" not in command
|
|
assert command[-2:] == ["--timeout", "900"]
|
|
|
|
|
|
def test_aider_command_adds_existing_routed_files(tmp_path) -> None:
|
|
git_dir = tmp_path / ".git"
|
|
git_dir.mkdir()
|
|
controller = tmp_path / "app/Http/Controllers/ChatController.php"
|
|
controller.parent.mkdir(parents=True)
|
|
controller.write_text("<?php\n")
|
|
|
|
command = AiderAgent(settings()).command(
|
|
git_dir / "task.txt",
|
|
("app/Http/Controllers/ChatController.php", "missing.php"),
|
|
)
|
|
|
|
assert ["--file", "app/Http/Controllers/ChatController.php"] == command[
|
|
command.index("--file"):command.index("--file") + 2
|
|
]
|
|
assert "missing.php" not in command
|
|
assert command[command.index("--map-tokens") + 1] == "0"
|
|
assert command[command.index("--edit-format") + 1] == "diff"
|
|
assert "--architect" not in command
|
|
assert "--editor-model" not in command
|
|
|
|
|
|
def test_unrouted_aider_command_stays_in_direct_edit_mode() -> None:
|
|
command = AiderAgent(settings()).command(Path("/repo/.git/task.txt"))
|
|
|
|
assert "--architect" not in command
|
|
|
|
|
|
def test_aider_detects_output_stall_at_configured_limit() -> None:
|
|
agent = AiderAgent(settings(agent_idle_timeout_seconds=120))
|
|
|
|
assert not agent._output_stalled(119.9)
|
|
assert agent._output_stalled(120)
|
|
|
|
|
|
def test_ollama_aider_command_uses_single_whole_file_pass(tmp_path) -> None:
|
|
git_dir = tmp_path / ".git"
|
|
git_dir.mkdir()
|
|
source = tmp_path / "config/erp_semantics.php"
|
|
source.parent.mkdir(parents=True)
|
|
source.write_text("<?php\n")
|
|
|
|
command = AiderAgent(settings(agent_model="ollama_chat/qwen3-coder:30b")).command(
|
|
git_dir / "task.txt",
|
|
("config/erp_semantics.php",),
|
|
)
|
|
|
|
assert command[command.index("--edit-format") + 1] == "whole"
|
|
assert "--no-auto-lint" in command
|
|
assert "--architect" not in command
|
|
assert "--editor-model" not in command
|
|
|
|
|
|
def test_ollama_test_only_correction_uses_precise_diff_format(tmp_path) -> None:
|
|
git_dir = tmp_path / ".git"
|
|
git_dir.mkdir()
|
|
test_file = tmp_path / "tests/Unit/FixTest.php"
|
|
test_file.parent.mkdir(parents=True)
|
|
test_file.write_text("<?php\n")
|
|
|
|
command = AiderAgent(settings(agent_model="ollama_chat/qwen3-coder:30b")).command(
|
|
git_dir / "task.txt",
|
|
("tests/Unit/FixTest.php",),
|
|
)
|
|
|
|
assert command[command.index("--edit-format") + 1] == "diff"
|
|
assert "--no-auto-lint" in command
|
|
|
|
|
|
def test_aider_timeout_has_actionable_error(tmp_path, monkeypatch) -> None:
|
|
workspace = tmp_path / "repo"
|
|
(workspace / ".git").mkdir(parents=True)
|
|
|
|
class TimedOutProcess:
|
|
pid = 4321
|
|
stdout = io.StringIO("partial output\n")
|
|
stderr = io.StringIO("")
|
|
returncode = None
|
|
|
|
def wait(self, timeout=None):
|
|
raise subprocess.TimeoutExpired("aider", timeout)
|
|
|
|
monkeypatch.setattr(
|
|
"lai_automation.agent.subprocess.Popen",
|
|
lambda *args, **kwargs: TimedOutProcess(),
|
|
)
|
|
signals = []
|
|
monkeypatch.setattr(
|
|
"lai_automation.agent.os.killpg",
|
|
lambda pid, sent_signal: signals.append((pid, sent_signal)),
|
|
)
|
|
monkeypatch.setattr("lai_automation.agent.os.chown", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr("lai_automation.agent.pwd.getpwnam", lambda _: type(
|
|
"Account", (), {"pw_dir": str(tmp_path), "pw_uid": 10001, "pw_gid": 10001}
|
|
)())
|
|
|
|
with pytest.raises(AgentTimeoutError, match="partial output"):
|
|
AiderAgent(settings(agent_timeout_seconds=60)).run(workspace, "fix it")
|
|
|
|
assert signals == [(4321, signal.SIGTERM), (4321, signal.SIGKILL)]
|
|
|
|
|
|
def test_aider_environment_does_not_include_database_credentials(tmp_path, monkeypatch) -> None:
|
|
config = settings(workspace_root=tmp_path)
|
|
workspace = tmp_path / "repo"
|
|
(workspace / ".git").mkdir(parents=True)
|
|
monkeypatch.setenv("TEST_DB_PASSWORD", "must-not-reach-aider")
|
|
monkeypatch.setenv("OPENROUTER_API_KEY", "qwen-provider-key")
|
|
monkeypatch.setenv("OLLAMA_API_BASE", "http://host.docker.internal:11434")
|
|
captured = {}
|
|
|
|
class SuccessfulProcess:
|
|
pid = 1234
|
|
stdout = io.StringIO("provider said qwen-provider-key\n")
|
|
stderr = io.StringIO("")
|
|
returncode = 0
|
|
|
|
def wait(self, timeout=None):
|
|
return self.returncode
|
|
|
|
def popen(command, **kwargs):
|
|
captured.update(kwargs["env"])
|
|
return SuccessfulProcess()
|
|
|
|
monkeypatch.setattr("lai_automation.agent.subprocess.Popen", popen)
|
|
monkeypatch.setattr("lai_automation.agent.os.chown", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr("lai_automation.agent.pwd.getpwnam", lambda _: type(
|
|
"Account", (), {"pw_dir": str(tmp_path), "pw_uid": 10001, "pw_gid": 10001}
|
|
)())
|
|
|
|
output = AiderAgent(config).run(workspace, "fix it")
|
|
|
|
assert "TEST_DB_PASSWORD" not in captured
|
|
assert "DB_PASSWORD" not in captured
|
|
assert captured["OPENROUTER_API_KEY"] == "qwen-provider-key"
|
|
assert captured["OLLAMA_API_BASE"] == "http://host.docker.internal:11434"
|
|
assert "qwen-provider-key" not in output
|
|
assert "[REDACTED]" in output
|
|
|
|
|
|
def test_successful_exit_with_only_provider_failure_is_rejected(tmp_path, monkeypatch) -> None:
|
|
workspace = tmp_path / "repo"
|
|
(workspace / ".git").mkdir(parents=True)
|
|
|
|
class ProviderFailedProcess:
|
|
pid = 1235
|
|
stdout = io.StringIO(
|
|
"litellm.InternalServerError: MistralException - Name or service not known\n"
|
|
"The API provider's servers are down or overloaded.\n"
|
|
)
|
|
stderr = io.StringIO("")
|
|
returncode = 0
|
|
|
|
def wait(self, timeout=None):
|
|
return self.returncode
|
|
|
|
monkeypatch.setattr(
|
|
"lai_automation.agent.subprocess.Popen",
|
|
lambda *args, **kwargs: ProviderFailedProcess(),
|
|
)
|
|
monkeypatch.setattr("lai_automation.agent.os.chown", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr("lai_automation.agent.pwd.getpwnam", lambda _: type(
|
|
"Account", (), {"pw_dir": str(tmp_path), "pw_uid": 10001, "pw_gid": 10001}
|
|
)())
|
|
|
|
agent = AiderAgent(settings())
|
|
with pytest.raises(AgentProviderError, match="provider failed"):
|
|
agent.run(workspace, "fix it")
|
|
|
|
assert agent.last_run_metadata["status"] == "provider_error" |