christian.meyer/lai_automatisierung
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: replace Aider with secure mini-SWE-agent worker
Some checks failed
CI / test (push) Has been cancelled
Details
CI / test (pull_request) Has been cancelled
Details

Browse Source
This commit is contained in:
Christian Meyer
2026-08-31 02:46:58 +02:00
parent 582db3a654
commit 25ed6a8439
30 changed files with 4215 additions and 2148 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
tests/conftest.py Normal file
View File

@@ -0,0 +1,45 @@
from __future__ import annotations
from pathlib import Path
import pytest
from lai_automation.config import Settings
@pytest.fixture
def settings(tmp_path: Path) -> Settings:
return Settings(
source_repository="https://gitea.example.test/owner/lai.git",
source_branch="main",
workspace_root=tmp_path,
gitea_url="https://gitea.example.test",
gitea_owner="owner",
gitea_repository="lai",
gitea_token="top-secret-token",
gitea_username="automation-bot",
queued_label="ai:queued",
running_label="ai:running",
ready_label="ai:ready",
failed_label="ai:failed",
poll_seconds=60,
worker_id="test-worker",
agent_model="mistral/devstral-latest",
agent_model_api_base="",
agent_command_timeout_seconds=180,
agent_wall_time_seconds=1200,
agent_step_limit=40,
agent_cost_limit=5.0,
agent_user="coder",
setup_commands=("composer install",),
test_commands=("php artisan test",),
max_changed_files=12,
max_changed_lines=800,
test_environment={
"APP_ENV": "testing",
"DB_DATABASE": "lai_agent_test",
"TEST_CUSTOMER_DB": "customer_agent_test",
"TEST_CONFIG_DB": "config_agent_test",
"DB_PASSWORD": "isolated-test-password",
},
)

78
tests/test_agent.py
View File

@@ -1,70 +1,20 @@
from pathlib import Path
from __future__ import annotations
from test_config import settings
from lai_automation.agent import AiderAgent
from lai_automation.agent import MiniSWEAgentRunner
from lai_automation.models import Issue
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 command.count("/dev/null") == 2
assert "--edit-format" not in command
assert command[-2:] == ["--timeout", "180"]
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"),
def test_task_uses_issue_text_not_github_parser() -> None:
issue = Issue(
number=42,
title="Wrong result",
body="Original prompt and expected result",
html_url="https://gitea.example.test/owner/lai/issues/42",
labels=(),
)
assert ["--file", "app/Http/Controllers/ChatController.php"] == command[
command.index("--file"):command.index("--file") + 2
]
assert "missing.php" not in command
assert "--architect" in command
assert command[command.index("--editor-model") + 1] == "mistral/test"
assert "--editor-edit-format" not in command
task = MiniSWEAgentRunner._task(issue)
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_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 = {}
def run(command, **kwargs):
captured.update(kwargs["env"])
return type("Result", (), {"returncode": 0, "stdout": "ok", "stderr": ""})()
monkeypatch.setattr("lai_automation.agent.subprocess.run", run)
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}
)())
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 "Issue #42: Wrong result" in task
assert issue.html_url in task
assert issue.body in task

117
tests/test_config.py
View File

@@ -1,75 +1,74 @@
from pathlib import Path
from __future__ import annotations
import pytest
from lai_automation.config import Settings
def settings(**overrides) -> Settings:
values = {
"database_host": "db",
"database_port": 3306,
"database_name": "lai",
"database_user": "worker",
"database_password": "secret",
"source_repository": "https://example.test/lai.git",
"source_branch": "main",
"workspace_root": Path("/tmp/workspaces"),
"mode": "dry-run",
"allow_direct_main": False,
"gitea_url": "https://example.test",
"gitea_owner": "owner",
"gitea_repository": "lai",
"gitea_token": "",
"gitea_username": "bot",
"agent_command": "aider",
"agent_model": "mistral/test",
"test_commands": ("php artisan test",),
"test_database_host": "test-db",
"test_database_port": 3306,
"test_database_name": "lai_agent_test",
"test_database_user": "test_runner",
"test_database_password": "test-secret",
"test_customer_database": "customer_agent_test",
"test_config_database": "cockpit_agent_test",
"poll_seconds": 60,
"max_changed_files": 12,
"max_changed_lines": 800,
"max_agent_attempts": 2,
"stale_claim_minutes": 90,
"dry_run": True,
}
values.update(overrides)
return Settings(**values)
def base_environment(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv(
"SOURCE_REPOSITORY",
"https://gitea.example.test/owner/lai.git",
)
monkeypatch.setenv("GITEA_URL", "https://gitea.example.test")
monkeypatch.setenv("GITEA_OWNER", "owner")
monkeypatch.setenv("GITEA_REPOSITORY", "lai")
monkeypatch.setenv("GITEA_TOKEN", "secret")
monkeypatch.setenv("MISTRAL_API_KEY", "model-secret")
monkeypatch.setenv("TEST_DB_NAME", "lai_agent_test")
monkeypatch.setenv("TEST_CUSTOMER_DB", "customer_agent_test")
monkeypatch.setenv("TEST_CONFIG_DB", "config_agent_test")
def test_direct_main_requires_explicit_kill_switch() -> None:
config = settings(mode="direct-main", dry_run=False, gitea_token="token")
with pytest.raises(ValueError, match="ALLOW_DIRECT_MAIN"):
config.validate()
def test_settings_load_issue_workflow_defaults(
monkeypatch: pytest.MonkeyPatch,
) -> None:
base_environment(monkeypatch)
settings = Settings.from_env()
assert settings.queued_label == "ai:queued"
assert settings.running_label == "ai:running"
assert settings.agent_model == "mistral/devstral-latest"
assert settings.test_commands == ("npm run build", "php artisan test")
assert settings.test_environment["DB_DATABASE"] == "lai_agent_test"
assert settings.test_environment["ERP_DATABASE"] == "customer_agent_test"
assert settings.test_environment["COCKPIT_CONFIG_DATABASE"] == "config_agent_test"
assert settings.test_environment["AGENT_TEST_DB_ENABLED"] == "true"
assert settings.test_environment["AI_MISTRAL_API_KEY"] == "test-only-placeholder"
assert "GITEA_TOKEN" not in settings.test_environment
assert "MISTRAL_API_KEY" not in settings.test_environment
def test_pr_requires_gitea_token() -> None:
config = settings(mode="pr", dry_run=False)
with pytest.raises(ValueError, match="GITEA_TOKEN"):
config.validate()
def test_settings_reject_non_https_source(monkeypatch: pytest.MonkeyPatch) -> None:
base_environment(monkeypatch)
monkeypatch.setenv(
"SOURCE_REPOSITORY",
"http://gitea.example.test/owner/lai.git",
)
with pytest.raises(ValueError, match="HTTPS"):
Settings.from_env()
def test_default_commands_initialize_disposable_laravel_schema(monkeypatch) -> None:
monkeypatch.setenv("FAILURE_DB_PASSWORD", "failure-secret")
monkeypatch.setenv("TEST_DB_PASSWORD", "test-secret")
monkeypatch.delenv("TEST_COMMANDS", raising=False)
def test_settings_reject_different_repository_host(
monkeypatch: pytest.MonkeyPatch,
) -> None:
base_environment(monkeypatch)
monkeypatch.setenv(
"SOURCE_REPOSITORY",
"https://attacker.example/owner/lai.git",
)
config = Settings.from_env()
assert "npm ci" in config.test_commands
assert "npm run build" in config.test_commands
assert "php artisan migrate:fresh --force" in config.test_commands
assert config.test_commands[-1] == "php artisan test"
with pytest.raises(ValueError, match="configured Gitea"):
Settings.from_env()
def test_agent_attempts_are_bounded() -> None:
with pytest.raises(ValueError, match="MAX_AGENT_ATTEMPTS"):
settings(max_agent_attempts=6).validate()
def test_settings_require_isolated_test_database(
monkeypatch: pytest.MonkeyPatch,
) -> None:
base_environment(monkeypatch)
monkeypatch.setenv("TEST_DB_NAME", "production")
settings(max_agent_attempts=5).validate()
with pytest.raises(ValueError, match="isolated test database"):
Settings.from_env()

107
tests/test_git.py
View File

@@ -1,107 +0,0 @@
import subprocess
from test_config import settings
from lai_automation.git import GitWorkspace
def test_test_command_does_not_interpret_shell_metacharacters(tmp_path) -> None:
workspace = GitWorkspace(settings(workspace_root=tmp_path), 1)
workspace.path.mkdir()
result = workspace.run("printf safe ; touch injected", check=False)
assert result.stdout == "safe"
assert result.returncode != 0
assert not (workspace.path / "injected").exists()
def test_summary_includes_untracked_files(tmp_path) -> None:
workspace = GitWorkspace(settings(workspace_root=tmp_path), 2)
workspace.path.mkdir()
subprocess.run(["git", "init"], cwd=workspace.path, check=True, capture_output=True)
(workspace.path / "new.php").write_text("<?php\n")
summary = workspace.summary()
assert summary.files == ("new.php",)
assert summary.added_lines == 1
def test_discard_changes_restores_clean_disposable_workspace(tmp_path) -> None:
workspace = GitWorkspace(settings(workspace_root=tmp_path), 3)
workspace.path.mkdir()
subprocess.run(["git", "init"], cwd=workspace.path, check=True, capture_output=True)
subprocess.run(
["git", "config", "user.email", "test@example.invalid"],
cwd=workspace.path,
check=True,
)
subprocess.run(
["git", "config", "user.name", "Test"],
cwd=workspace.path,
check=True,
)
tracked = workspace.path / "tracked.php"
tracked.write_text("<?php\nreturn [];\n")
subprocess.run(["git", "add", "."], cwd=workspace.path, check=True)
subprocess.run(["git", "commit", "-m", "base"], cwd=workspace.path, check=True, capture_output=True)
tracked.write_text("<?php\ninvalid\n")
untracked = workspace.path / "extra.php"
untracked.write_text("<?php\n")
workspace.discard_changes()
assert tracked.read_text() == "<?php\nreturn [];\n"
assert not untracked.exists()
def test_checkpoint_restores_last_valid_changes(tmp_path) -> None:
workspace = GitWorkspace(settings(workspace_root=tmp_path), 4)
workspace.path.mkdir()
subprocess.run(["git", "init"], cwd=workspace.path, check=True, capture_output=True)
subprocess.run(
["git", "config", "user.email", "test@example.invalid"],
cwd=workspace.path,
check=True,
)
subprocess.run(
["git", "config", "user.name", "Test"],
cwd=workspace.path,
check=True,
)
tracked = workspace.path / "config.php"
tracked.write_text("<?php\nreturn [];\n")
subprocess.run(["git", "add", "."], cwd=workspace.path, check=True)
subprocess.run(["git", "commit", "-m", "base"], cwd=workspace.path, check=True, capture_output=True)
tracked.write_text("<?php\nreturn ['valid'];\n")
added = workspace.path / "test.php"
added.write_text("<?php\n")
checkpoint = workspace.checkpoint()
tracked.write_text("<?php\nreturn [broken;\n")
workspace.restore_checkpoint(checkpoint)
assert tracked.read_text() == "<?php\nreturn ['valid'];\n"
assert added.exists()
def test_diff_returns_discarded_patch_excerpt(tmp_path) -> None:
workspace = GitWorkspace(settings(workspace_root=tmp_path), 5)
workspace.path.mkdir()
subprocess.run(["git", "init"], cwd=workspace.path, check=True, capture_output=True)
tracked = workspace.path / "service.php"
tracked.write_text("<?php\nreturn 'old';\n")
subprocess.run(["git", "add", "."], cwd=workspace.path, check=True)
subprocess.run(
["git", "-c", "user.email=test@example.invalid", "-c", "user.name=Test", "commit", "-m", "base"],
cwd=workspace.path,
check=True,
capture_output=True,
)
tracked.write_text("<?php\nreturn 'new';\n")
patch = workspace.diff()
assert "-return 'old';" in patch
assert "+return 'new';" in patch

60
tests/test_git_workspace.py Normal file
View File

@@ -0,0 +1,60 @@
import subprocess
from dataclasses import replace
from pathlib import Path
from unittest.mock import Mock
import pytest
from lai_automation.git import GitWorkspace
from lai_automation.models import ChangeSummary
def test_check_diff_accepts_clean_changed_file(settings, tmp_path: Path) -> None:
workspace = GitWorkspace(replace(settings, workspace_root=tmp_path), 1)
workspace.path.mkdir(parents=True)
(workspace.path / "app.php").write_text("<?php\nreturn true;\n")
workspace._git = Mock(
side_effect=[
subprocess.CompletedProcess([], 0, "", ""),
subprocess.CompletedProcess([], 0, "app.php\n", ""),
]
)
workspace.check_diff(ChangeSummary(("app.php",), 2, 0))
assert workspace._git.call_count == 2
def test_check_diff_rejects_trailing_whitespace_in_untracked_file(
settings,
tmp_path: Path,
) -> None:
workspace = GitWorkspace(replace(settings, workspace_root=tmp_path), 1)
workspace.path.mkdir(parents=True)
(workspace.path / "app.php").write_text("<?php \n")
workspace._git = Mock(
side_effect=[
subprocess.CompletedProcess([], 0, "", ""),
subprocess.CompletedProcess([], 0, "app.php\n", ""),
]
)
with pytest.raises(RuntimeError, match="trailing whitespace"):
workspace.check_diff(ChangeSummary(("app.php",), 1, 0))
def test_check_diff_ignores_unchanged_whitespace_in_tracked_file(
settings,
tmp_path: Path,
) -> None:
workspace = GitWorkspace(replace(settings, workspace_root=tmp_path), 1)
workspace.path.mkdir(parents=True)
(workspace.path / "legacy.php").write_text("<?php \nreturn true;\n")
workspace._git = Mock(
side_effect=[
subprocess.CompletedProcess([], 0, "", ""),
subprocess.CompletedProcess([], 0, "", ""),
]
)
workspace.check_diff(ChangeSummary(("legacy.php",), 1, 0))

139
tests/test_gitea.py
View File

@@ -1,27 +1,130 @@
from __future__ import annotations
from unittest.mock import Mock
import pytest
from test_config import settings
from lai_automation.gitea import GiteaClient
from lai_automation.models import Issue, Label
def test_create_pull_request_requires_url() -> None:
client = GiteaClient(settings(gitea_token="token"))
response = Mock()
response.json.return_value = {"number": 12}
client.session.post = Mock(return_value=response)
with pytest.raises(ValueError, match="pull request URL"):
client.create_pull_request("ai/failure-1", "Fix", "Body")
def response(payload, status: int = 200) -> Mock:
result = Mock()
result.json.return_value = payload
result.status_code = status
result.text = ""
result.raise_for_status.return_value = None
return result
def test_create_pull_request_returns_html_url() -> None:
client = GiteaClient(settings(gitea_token="token"))
response = Mock()
response.json.return_value = {"html_url": "https://example.test/owner/lai/pulls/12"}
client.session.post = Mock(return_value=response)
def test_list_queued_issues_filters_pull_requests(settings) -> None:
client = GiteaClient(settings)
client.session.request = Mock(
return_value=response(
[
{
"number": 7,
"title": "Wrong answer",
"body": "Details",
"html_url": "https://gitea.example.test/owner/lai/issues/7",
"labels": [{"id": 1, "name": "ai:queued"}],
},
{
"number": 8,
"title": "A pull request",
"body": "",
"html_url": "https://gitea.example.test/owner/lai/pulls/8",
"labels": [{"id": 1, "name": "ai:queued"}],
"pull_request": {"merged": False},
},
]
)
)
result = client.create_pull_request("ai/failure-1", "Fix", "Body")
issues = client.list_queued_issues()
assert result.endswith("/pulls/12")
assert [issue.number for issue in issues] == [7]
assert client.session.request.call_args.kwargs["params"]["labels"] == "ai:queued"
def test_ensure_automation_labels_creates_only_missing_labels(settings) -> None:
client = GiteaClient(settings)
existing = [{"id": 1, "name": "ai:queued"}]
created = iter(
[
{"id": 2, "name": "ai:running"},
{"id": 3, "name": "ai:ready"},
{"id": 4, "name": "ai:failed"},
]
)
def request(method, _url, **kwargs):
if method == "GET":
return response(existing)
return response(next(created))
client.session.request = Mock(side_effect=request)
labels = client.ensure_automation_labels()
assert set(labels) == {"ai:queued", "ai:running", "ai:ready", "ai:failed"}
assert client.session.request.call_count == 4
def test_state_transition_preserves_manual_labels(settings) -> None:
client = GiteaClient(settings)
client.session.request = Mock(
side_effect=[
response(
[
{"id": 9, "name": "bug"},
{"id": 2, "name": "ai:running"},
]
),
response(
{
"number": 5,
"title": "Issue",
"body": "Body",
"html_url": "https://gitea.example.test/owner/lai/issues/5",
"labels": [
{"id": 9, "name": "bug"},
{"id": 2, "name": "ai:running"},
],
}
),
]
)
issue = Issue(
number=5,
title="Issue",
body="Body",
html_url="https://gitea.example.test/owner/lai/issues/5",
labels=(Label(9, "bug"), Label(1, "ai:queued")),
)
known = {
"ai:queued": Label(1, "ai:queued"),
"ai:running": Label(2, "ai:running"),
}
updated = client.set_automation_state(issue, "ai:running", known)
assert updated.has_label("ai:running")
replace_call = client.session.request.call_args_list[0]
assert replace_call.args[0] == "PUT"
assert replace_call.args[1].endswith("/issues/5/labels")
assert replace_call.kwargs["json"] == {"labels": [9, 2]}
assert client.session.request.call_args_list[1].args[0] == "GET"
def test_create_pull_request_returns_number_and_url(settings) -> None:
client = GiteaClient(settings)
client.session.request = Mock(
return_value=response(
{"number": 17, "html_url": "https://gitea.example.test/owner/lai/pulls/17"}
)
)
pull_request = client.create_pull_request("ai/issue-5", "Fix", "Fixes #5")
assert pull_request.number == 17
assert pull_request.html_url.endswith("/pulls/17")
assert client.session.request.call_args.kwargs["json"]["base"] == "main"

136
tests/test_policy.py
View File

@@ -1,85 +1,83 @@
from dataclasses import replace
from __future__ import annotations
from test_config import settings
from unittest.mock import Mock
from lai_automation.git import ChangeSummary
from lai_automation.models import FailureReport
from lai_automation.policy import evaluate_changes, evaluate_report_scope
import pytest
from lai_automation.models import ChangeSummary
from lai_automation.policy import validate_changes
def test_policy_accepts_small_application_change() -> None:
result = evaluate_changes(
ChangeSummary(("app/Services/Example.php", "tests/Feature/ExampleTest.php"), 40, 5),
settings(),
)
assert result.allowed
def workspace_with_no_symlinks() -> Mock:
workspace = Mock()
workspace.has_symlink.return_value = False
workspace.has_path.return_value = False
return workspace
def test_policy_rejects_secrets_and_vendor_paths() -> None:
result = evaluate_changes(
ChangeSummary((".env", "vendor/foo.php", ".aider.tags.cache.v4/cache.db"), 3, 0),
settings(),
)
assert not result.allowed
assert len(result.errors) == 3
def test_policy_rejects_test_only_repair() -> None:
result = evaluate_changes(
ChangeSummary(("tests/Unit/ExampleTest.php",), 20, 2),
settings(),
def test_accepts_small_application_change(settings) -> None:
summary = ChangeSummary(
("app/Services/AIService.php", "tests/Feature/AITest.php"),
25,
4,
)
assert not result.allowed
assert "production behavior was not fixed" in result.errors[0]
validate_changes(workspace_with_no_symlinks(), summary, settings)
def test_direct_main_blocks_validator_and_migrations() -> None:
config = replace(
settings(),
mode="direct-main",
dry_run=False,
allow_direct_main=True,
gitea_token="token",
)
result = evaluate_changes(
ChangeSummary(("app/Services/SQLValidatorService.php", "database/migrations/new.php"), 10, 2),
config,
)
assert not result.allowed
@pytest.mark.parametrize(
"filename",
[
".env",
".env.example",
".gitea/workflows/ci.yml",
"docker-compose.yml",
"docker/php/Dockerfile",
"storage/logs/output.log",
"vendor/package/file.php",
],
)
def test_rejects_protected_paths(settings, filename: str) -> None:
summary = ChangeSummary((filename, "app/Fix.php"), 2, 0)
with pytest.raises(ValueError, match="Protected"):
validate_changes(workspace_with_no_symlinks(), summary, settings)
def test_policy_rejects_symlinks(tmp_path) -> None:
(tmp_path / "link.php").symlink_to("/etc/passwd")
result = evaluate_changes(ChangeSummary(("link.php",), 1, 0), settings(), tmp_path)
assert not result.allowed
def test_rejects_git_metadata_created_in_worktree(settings) -> None:
workspace = workspace_with_no_symlinks()
workspace.has_path.return_value = True
with pytest.raises(ValueError, match="Git metadata"):
validate_changes(
workspace,
ChangeSummary(("app/Fix.php",), 1, 0),
settings,
)
def test_cross_database_report_requires_application_orchestration() -> None:
report = FailureReport(
id=56,
chat_message_id=1,
source="user_feedback",
category="incorrect_answer",
user_feedback="Use cockpit_config.time_registration",
feedback_type="wrong_result",
expected_behavior="Join cockpit_config manager_id and FK_AuftragPos",
prompt="Hours in April",
response="0 hours",
sql_query="SELECT 0",
error_message=None,
context={},
def test_rejects_tests_without_application_change(settings) -> None:
summary = ChangeSummary(("tests/Feature/FixTest.php",), 20, 0)
with pytest.raises(ValueError, match="application source"):
validate_changes(workspace_with_no_symlinks(), summary, settings)
def test_rejects_symlink(settings) -> None:
workspace = workspace_with_no_symlinks()
workspace.has_symlink.return_value = True
summary = ChangeSummary(("app/Fix.php",), 1, 0)
with pytest.raises(ValueError, match="Symbolic links"):
validate_changes(workspace, summary, settings)
def test_rejects_oversized_diff(settings) -> None:
summary = ChangeSummary(
("app/Fix.php",),
settings.max_changed_lines + 1,
0,
)
rejected = evaluate_report_scope(
report,
ChangeSummary(("config/erp_semantics.php", "tests/Unit/FixTest.php"), 20, 2),
)
accepted = evaluate_report_scope(
report,
ChangeSummary(("app/Http/Controllers/ChatController.php", "tests/Unit/FixTest.php"), 20, 2),
)
assert not rejected.allowed
assert "staged orchestration" in rejected.errors[0]
assert accepted.allowed
with pytest.raises(ValueError, match="line limit"):
validate_changes(workspace_with_no_symlinks(), summary, settings)

210
tests/test_prompt.py
View File

@@ -1,210 +0,0 @@
from lai_automation.models import FailureReport
from lai_automation.prompt import build_retry_task, build_task, redact, repair_files
def test_redact_removes_common_secrets_and_personal_markers() -> None:
value = "password=hunter2 token: abc123 admin@example.com 10.2.3.4 Bearer secret.jwt"
result = redact(value)
assert "hunter2" not in result
assert "abc123" not in result
assert "admin@example.com" not in result
assert "10.2.3.4" not in result
assert "secret.jwt" not in result
def test_task_contains_constraints_and_redacted_report() -> None:
report = FailureReport(
id=7,
chat_message_id=91,
source="user_feedback",
category="incorrect_answer",
user_feedback="password=secret-value",
feedback_type="wrong_join",
expected_behavior="Join FK_Artikel with FK_AuftragPos and filter Auftragkennung = 3.",
prompt="List locations for admin@example.com",
response="The wrong product count was returned.",
sql_query="SELECT * FROM company",
error_message=None,
context={"model": "mistral", "database_type": "customer"},
)
task = build_task(report)
assert "failure report #7" in task
assert "secret-value" not in task
assert "admin@example.com" not in task
assert "Do not commit or push" in task
assert "never mock or stub the behavior being tested" in task
assert "Repository ownership map:" in task
assert "config/erp_semantics.php owns ERP joins" in task
assert "Tests must contain assertions that execute" in task
assert "Do not expose protected production methods publicly" in task
assert "Feedback type: wrong_join" in task
assert "Join FK_Artikel with FK_AuftragPos and filter Auftragkennung = 3." in task
assert "The wrong product count was returned." in task
def test_redact_removes_extended_credentials() -> None:
private_key = "-----BEGIN PRIVATE KEY-----\nprivate-material\n-----END PRIVATE KEY-----"
value = (
"AKIAIOSFODNN7EXAMPLE "
"eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0In0.signature "
"mysql://admin:database-password@db.internal/lai "
f"{private_key}"
)
result = redact(value)
assert "AKIAIOSFODNN7EXAMPLE" not in result
assert "eyJhbGci" not in result
assert "database-password" not in result
assert "private-material" not in result
def test_retry_task_contains_failure_without_secrets() -> None:
report = FailureReport(
id=8,
chat_message_id=1,
source="user_feedback",
category="incorrect_answer",
user_feedback="Wrong result",
feedback_type="wrong_result",
expected_behavior="Use a weighted average",
prompt="Average sale price",
response="Zero",
sql_query="SELECT AVG(list_price)",
error_message=None,
context={"database_type": "erp"},
)
task = build_retry_task(report, RuntimeError("Missing import; token=secret-value"))
assert "Missing import" in task
assert "secret-value" not in task
assert "Do not remove or weaken tests" in task
assert "unrelated production routing" in task
assert "Use a weighted average" in task
assert "SELECT AVG(list_price)" in task
def test_incorrect_answer_task_guides_semantics_level_repair() -> None:
report = FailureReport(
id=42,
chat_message_id=1,
source="user_feedback",
category="incorrect_answer",
user_feedback="Wrong result",
feedback_type="wrong_result",
expected_behavior="Sum net values and divide by quantity",
prompt="Average sale price for article 9031",
response="0.00",
sql_query="SELECT AVG(Vk_preis) FROM FK_Artikel",
error_message=None,
context={"database_type": "erp"},
)
task = build_task(report)
assert "config/*_semantics.php" in task
assert "Do not hardcode" in task
assert "protected schema/business-" in task
assert "context builder" in task
assert "known-bad SQL" in task
assert "compare every alias.column" in task
assert "Never replace one missing column" in task
def test_retry_task_keeps_failure_from_start_of_long_output() -> None:
report = FailureReport(
id=43,
chat_message_id=1,
source="automatic",
category="sql_execution",
user_feedback=None,
feedback_type=None,
expected_behavior=None,
prompt="Top products",
response=None,
sql_query="SELECT a.missing FROM orders a",
error_message="Unknown column",
context={"database_type": "erp"},
)
task = build_retry_task(report, RuntimeError("DECISIVE ASSERTION\n" + "x" * 11_000))
assert "DECISIVE ASSERTION" in task
def test_customer_database_task_explains_tenant_scope() -> None:
report = FailureReport(
id=13,
chat_message_id=1,
source="automatic",
category="sql_execution",
user_feedback=None,
feedback_type=None,
expected_behavior=None,
prompt="What did this customer buy most?",
response=None,
sql_query="SELECT * FROM invoice_row WHERE InvoiceCreator = 'tenant'",
error_message="Unknown column InvoiceCreator",
context={"database_type": "customer"},
)
task = build_task(report)
assert "already selected one tenant database" in task
assert "must not" in task
assert "guessed column" in task
assert "tenant_type" in task
def test_unresolved_placeholder_task_guides_real_answer_path() -> None:
report = FailureReport(
id=49,
chat_message_id=1,
source="user_feedback",
category="incorrect_answer",
user_feedback="Token was not restored",
feedback_type="wrong_result",
expected_behavior="Return the database value",
prompt="Average price",
response="Price: __LAI_UNKNOWN_VALUE__",
sql_query="SELECT AVG(price)",
error_message=None,
context={"database_type": "erp"},
)
task = build_task(report)
assert "AIService::generateDatabaseResponse" in task
assert "exact token map" in task
assert "HTTP-backed AIService test" in task
assert "local database-result fallback" in task
def test_cross_database_report_guides_staged_source_routing() -> None:
report = FailureReport(
id=56,
chat_message_id=1,
source="user_feedback",
category="incorrect_answer",
user_feedback="Wrong hours",
feedback_type="wrong_result",
expected_behavior="Join cockpit_config.time_registration to customer manager and ERP positions",
prompt="Hours in April",
response="0 hours",
sql_query="SELECT SUM(hours) FROM LX_AUFGABE",
error_message=None,
context={"database_type": "erp"},
)
task = build_task(report)
assert "tables from multiple databases" in task
assert "staged-query workflows" in task
assert "Do not put an impossible cross-connection join" in task
assert "preserving tenant isolation" in task
assert "editing only config/*_semantics.php or tests is an invalid repair" in task
assert "describe the data" in task
assert "flow, connection used at each stage" in task
assert repair_files(report) == (
"app/Http/Controllers/ChatController.php",
"app/Services/TenantDatabaseService.php",
"tests/Unit/ChatControllerSourceResolutionTest.php",
)

111
tests/test_security.py Normal file
View File

@@ -0,0 +1,111 @@
from __future__ import annotations
from pathlib import Path
import pytest
from minisweagent.exceptions import Submitted
from lai_automation.security import SafeLocalEnvironment, redact
def test_environment_does_not_inherit_worker_secrets(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("GITEA_TOKEN", "gitea-secret")
monkeypatch.setenv("MISTRAL_API_KEY", "model-secret")
environment = SafeLocalEnvironment(
cwd=str(tmp_path),
env={"APP_ENV": "testing"},
username=None,
)
result = environment.execute(
{"command": "printf '%s|%s|%s' \"$GITEA_TOKEN\" \"$MISTRAL_API_KEY\" \"$APP_ENV\""}
)
assert result["output"] == "||testing"
assert "GITEA_TOKEN" not in environment.get_template_vars()
assert "MISTRAL_API_KEY" not in environment.get_template_vars()
assert "gitea-secret" not in str(environment.serialize())
def test_environment_rejects_cwd_outside_repository(tmp_path: Path) -> None:
repository = tmp_path / "repo"
repository.mkdir()
environment = SafeLocalEnvironment(cwd=str(repository), username=None)
with pytest.raises(ValueError, match="inside the repository"):
environment.execute({"command": "pwd"}, cwd=str(tmp_path))
@pytest.mark.parametrize(
"command",
[
"git status",
"cat /proc/self/environ",
"cat .env",
"touch /tmp/agent-file",
"cd .. && pwd",
"php -S 127.0.0.1:8000",
"php artisan serve &",
],
)
def test_environment_rejects_unsafe_commands(tmp_path: Path, command: str) -> None:
environment = SafeLocalEnvironment(cwd=str(tmp_path), username=None)
result = environment.execute({"command": command})
assert result["returncode"] == -1
assert "safety policy" in result["exception_info"]
assert not (tmp_path / "agent-file").exists()
def test_environment_allows_commands_within_repository(tmp_path: Path) -> None:
nested = tmp_path / "tests"
nested.mkdir()
environment = SafeLocalEnvironment(cwd=str(tmp_path), username=None)
result = environment.execute(
{"command": f"cd {nested} && printf focused-test"}
)
assert result["returncode"] == 0
assert result["output"] == "focused-test"
def test_environment_allows_standard_error_redirection(tmp_path: Path) -> None:
environment = SafeLocalEnvironment(cwd=str(tmp_path), username=None)
result = environment.execute({"command": "printf redirected 2>&1"})
assert result["returncode"] == 0
assert result["output"] == "redirected"
def test_environment_recognizes_submission_marker(tmp_path: Path) -> None:
environment = SafeLocalEnvironment(cwd=str(tmp_path), username=None)
with pytest.raises(Submitted) as submission:
environment.execute(
{"command": "printf 'COMPLETE_TASK_AND_SUBMIT_FINAL_OUTPUT\\ndone\\n'"}
)
assert submission.value.messages[0]["extra"]["submission"] == "done\n"
@pytest.mark.parametrize(
"value",
[
"Authorization: Bearer abcdefghijklmnop",
"api_key=abcdefghijklmnop",
"password: supersecret",
"https://bot:supersecret@gitea.example.test/repo.git",
"sk-abcdefghijklmnop",
],
)
def test_redact_removes_common_secret_forms(value: str) -> None:
result = redact(value)
assert "supersecret" not in result
assert "abcdefghijklmnop" not in result
assert "[REDACTED]" in result

518
tests/test_worker.py
View File

@@ -1,351 +1,195 @@
import threading
from pathlib import Path
from unittest.mock import MagicMock, Mock, patch
from __future__ import annotations
from test_config import settings
from unittest.mock import Mock, call, patch
from lai_automation.git import ChangeSummary
from lai_automation.models import FailureReport
from lai_automation.worker import AutomationWorker
from lai_automation.models import (
AgentResult,
ChangeSummary,
Issue,
Label,
PullRequest,
)
from lai_automation.worker import Worker
def test_dry_run_generates_task_without_agent_or_git() -> None:
report = FailureReport(
id=11,
chat_message_id=1,
source="automatic",
category="sql_execution",
user_feedback=None,
feedback_type=None,
expected_behavior=None,
prompt="List locations",
response=None,
sql_query=None,
error_message="Unknown column",
context={},
def queued_issue() -> Issue:
return Issue(
number=7,
title="Wrong calculation",
body="The original report",
html_url="https://gitea.example.test/owner/lai/issues/7",
labels=(Label(1, "ai:queued"), Label(9, "bug")),
)
worker = AutomationWorker(settings())
worker.reports = Mock()
worker.reports.claim_next.return_value = report
assert worker.run_once()
worker.reports.release.assert_called_once()
worker.agent = Mock()
worker.agent.run.assert_not_called()
def test_stop_signal_interrupts_worker_wait() -> None:
worker = AutomationWorker(settings())
worker._stop(15, None)
assert worker.stop_event.is_set()
def labels() -> dict[str, Label]:
return {
"ai:queued": Label(1, "ai:queued"),
"ai:running": Label(2, "ai:running"),
"ai:ready": Label(3, "ai:ready"),
"ai:failed": Label(4, "ai:failed"),
}
def test_database_credentials_are_only_passed_to_test_process() -> None:
worker = AutomationWorker(settings())
worker._prepare_test_schema = Mock()
workspace = Mock()
workspace.run.return_value = Mock(returncode=0, stdout="ok", stderr="")
results = worker._run_tests(workspace)
assert results[0]["returncode"] == 0
environment = workspace.run.call_args.kwargs["extra_environment"]
assert environment["AGENT_TEST_DB_ENABLED"] == "true"
assert environment["APP_ENV"] == "testing"
assert environment["AI_MISTRAL_API_KEY"] == "automation-test-key"
assert environment["TEST_DB_USER"] == "test_runner"
assert environment["TEST_DB_PASSWORD"] == "test-secret"
assert environment["DB_DATABASE"] == "lai_agent_test"
assert environment["ERP_DATABASE"] == "customer_agent_test"
assert environment["COCKPIT_CONFIG_DATABASE"] == "cockpit_agent_test"
def test_test_schema_creates_minimal_erp_lookup(monkeypatch) -> None:
worker = AutomationWorker(settings())
cursor = Mock()
connection = MagicMock()
connection.cursor.return_value.__enter__.return_value = cursor
connect = Mock(return_value=connection)
monkeypatch.setattr("lai_automation.worker.pymysql.connect", connect)
worker._prepare_test_schema()
assert connect.call_args.kwargs["database"] == "customer_agent_test"
assert "CREATE TABLE IF NOT EXISTS FK_Nummerkreis" in cursor.execute.call_args.args[0]
assert cursor.executemany.call_args.args[1][0] == (3, "Rechnung", "RG")
connection.close.assert_called_once()
def test_validation_commands_are_logged_without_output(caplog) -> None:
worker = AutomationWorker(settings())
worker._prepare_test_schema = Mock()
workspace = Mock()
workspace.run.return_value = Mock(returncode=0, stdout="sensitive output", stderr="")
with caplog.at_level("INFO"):
worker._run_tests(workspace, report_id=7, attempt=2)
assert "Report 7 attempt 2 validation command: php artisan test starting" in caplog.text
assert "returncode=0" in caplog.text
assert "sensitive output" not in caplog.text
def test_failed_validation_preserves_start_and_end_of_long_output() -> None:
worker = AutomationWorker(settings())
worker._prepare_test_schema = Mock()
workspace = Mock()
workspace.run.return_value = Mock(
returncode=1,
stdout="DECISIVE ASSERTION\n" + "x" * 13_000,
stderr="FINAL ERROR",
@patch("lai_automation.worker.GitWorkspace")
def test_successful_issue_creates_pr_and_ready_label(
GitWorkspace,
settings,
) -> None:
issue = queued_issue()
running = Issue(
**{
**issue.__dict__,
"labels": (Label(2, "ai:running"), Label(9, "bug")),
}
)
gitea = Mock()
gitea.get_issue.return_value = issue
gitea.set_automation_state.side_effect = [running, running]
gitea.create_pull_request.return_value = PullRequest(
18,
"https://gitea.example.test/owner/lai/pulls/18",
)
agent = Mock()
agent.run.return_value = AgentResult(
"Submitted",
"Implemented and tested",
4,
0.2,
)
workspace = GitWorkspace.return_value
workspace.path = settings.workspace_root / "issue-7" / "repo"
workspace.branch = "ai/issue-7"
workspace.summary.return_value = ChangeSummary(
("app/Services/Fix.php", "tests/Feature/FixTest.php"),
20,
2,
)
workspace.has_symlink.return_value = False
workspace.has_path.return_value = False
workspace.commit.return_value = "abc123"
Worker(settings, gitea, agent).process(issue, labels())
workspace.prepare.assert_called_once()
workspace.run_trusted.assert_has_calls(
[
call(
"composer install",
extra_environment=settings.test_environment,
),
call(
"php artisan test",
extra_environment=settings.test_environment,
),
]
)
workspace.push.assert_called_once()
assert workspace.check_diff.call_count == 2
assert gitea.create_pull_request.call_args.args[0] == "ai/issue-7"
assert "Fixes #7" in gitea.create_pull_request.call_args.args[2]
assert "The original report" in gitea.create_pull_request.call_args.args[2]
assert gitea.set_automation_state.call_args_list[-1].args[1] == "ai:ready"
workspace.cleanup.assert_called_once()
@patch("lai_automation.worker.GitWorkspace")
def test_failed_tests_prevent_push_and_mark_issue_failed(
GitWorkspace,
settings,
) -> None:
issue = queued_issue()
running = Issue(
**{
**issue.__dict__,
"labels": (Label(2, "ai:running"),),
}
)
gitea = Mock()
gitea.get_issue.side_effect = [issue, running]
gitea.set_automation_state.return_value = running
agent = Mock()
agent.run.return_value = AgentResult("Submitted", "Done", 2, 0.1)
workspace = GitWorkspace.return_value
workspace.path = settings.workspace_root / "issue-7" / "repo"
workspace.summary.return_value = ChangeSummary(("app/Fix.php",), 2, 0)
workspace.has_symlink.return_value = False
workspace.has_path.return_value = False
workspace.run_trusted.side_effect = [
Mock(),
RuntimeError("tests failed with token=hidden"),
]
Worker(settings, gitea, agent).process(issue, labels())
workspace.push.assert_not_called()
gitea.create_pull_request.assert_not_called()
assert gitea.set_automation_state.call_args_list[-1].args[1] == "ai:failed"
failure_comment = gitea.comment.call_args_list[-1].args[1]
assert "tests failed" in failure_comment
assert "hidden" not in failure_comment
def test_run_once_only_processes_first_queued_issue(settings) -> None:
gitea = Mock()
issue = queued_issue()
second_issue = Issue(**{**issue.__dict__, "number": 8})
expected_labels = labels()
gitea.ensure_automation_labels.return_value = expected_labels
gitea.list_queued_issues.return_value = [issue, second_issue]
worker = Worker(settings, gitea, Mock())
worker.process = Mock()
assert worker.run_once() is True
worker.process.assert_called_once_with(issue, expected_labels)
def test_run_once_returns_false_without_work(settings) -> None:
gitea = Mock()
gitea.ensure_automation_labels.return_value = labels()
gitea.list_queued_issues.return_value = []
assert Worker(settings, gitea, Mock()).run_once() is False
@patch("lai_automation.worker.GitWorkspace")
def test_preflight_runs_setup_and_tests_without_model(
GitWorkspace,
settings,
) -> None:
workspace = GitWorkspace.return_value
agent = Mock()
Worker(settings, Mock(), agent).preflight()
GitWorkspace.assert_called_once_with(settings, 0)
workspace.prepare.assert_called_once()
workspace.run_trusted.assert_has_calls(
[
call(
"composer install",
extra_environment=settings.test_environment,
),
call(
"php artisan test",
extra_environment=settings.test_environment,
),
]
)
agent.run.assert_not_called()
workspace.cleanup.assert_called_once()
@patch("lai_automation.worker.GitWorkspace")
def test_preflight_cleans_up_after_failure(GitWorkspace, settings) -> None:
workspace = GitWorkspace.return_value
workspace.run_trusted.side_effect = RuntimeError("test failure")
try:
worker._run_tests(workspace)
except RuntimeError as exception:
failure = str(exception)
Worker(settings, Mock(), Mock()).preflight()
except RuntimeError:
pass
else:
raise AssertionError("Expected failed validation")
raise AssertionError("Expected preflight failure")
assert "DECISIVE ASSERTION" in failure
assert "...[middle omitted]..." in failure
assert "FINAL ERROR" in failure
def test_long_stage_emits_heartbeat(caplog) -> None:
worker = AutomationWorker(settings())
with caplog.at_level("INFO"), worker._progress(7, "agent attempt 1", interval=0.001):
threading.Event().wait(0.01)
assert "Report 7 agent attempt 1 still running" in caplog.text
assert "Report 7 agent attempt 1 finished" in caplog.text
def test_php_syntax_failure_discards_attempt_before_full_validation() -> None:
report = FailureReport(
id=56,
chat_message_id=1,
source="user_feedback",
category="incorrect_answer",
user_feedback="Wrong hours",
feedback_type="wrong_result",
expected_behavior="Use time registration",
prompt="Hours in April",
response="0 hours",
sql_query="SELECT 0",
error_message=None,
context={},
)
config = settings(mode="pr", dry_run=False, gitea_token="token")
worker = AutomationWorker(config)
worker.reports = Mock()
worker.reports.claim_next.return_value = report
worker.agent = Mock()
worker.agent.run.side_effect = ["broken PHP", "fixed PHP"]
worker.gitea = Mock()
worker._run_tests = Mock(return_value=[])
workspace = Mock()
workspace.path = Path("/tmp/nonexistent-workspace")
workspace.branch = "ai/failure-56"
workspace.summary.return_value = ChangeSummary(
("config/erp_semantics.php", "tests/Unit/FixTest.php"), 20, 2
)
workspace.run_args.side_effect = [
Mock(returncode=255, stdout="Parse error", stderr="unexpected token"),
Mock(returncode=0, stdout="No syntax errors", stderr=""),
Mock(returncode=0, stdout="No syntax errors", stderr=""),
]
workspace.diff.return_value = "--- app/Service.php\n+++ app/Service.php\n+intended fix"
workspace.checkpoint.return_value = "valid-checkpoint"
workspace.commit.return_value = "commit-id"
with patch("lai_automation.worker.GitWorkspace", return_value=workspace):
assert worker.run_once()
assert worker.agent.run.call_count == 2
assert "unexpected token" in worker.agent.run.call_args_list[1].args[1]
assert "clean repository" in worker.agent.run.call_args_list[1].args[1]
assert "intended fix" in worker.agent.run.call_args_list[1].args[1]
assert "repair-only attempt" in worker.agent.run.call_args_list[1].args[1]
workspace.restore_checkpoint.assert_called_once_with(None)
worker._run_tests.assert_called_once()
def test_malformed_retry_restores_last_syntax_valid_attempt() -> None:
report = FailureReport(
id=56,
chat_message_id=1,
source="user_feedback",
category="incorrect_answer",
user_feedback="Wrong hours",
feedback_type="wrong_result",
expected_behavior="Use time registration",
prompt="Hours in April",
response="0 hours",
sql_query="SELECT 0",
error_message=None,
context={},
)
config = settings(mode="pr", dry_run=False, gitea_token="token", max_agent_attempts=3)
worker = AutomationWorker(config)
worker.reports = Mock()
worker.reports.claim_next.return_value = report
worker.agent = Mock()
worker.agent.run.side_effect = ["valid but failing", "broken retry", "corrected retry"]
worker.gitea = Mock()
worker._run_tests = Mock(side_effect=[RuntimeError("Expected hour rule"), []])
workspace = Mock()
workspace.path = Path("/tmp/nonexistent-workspace")
workspace.branch = "ai/failure-56"
workspace.summary.return_value = ChangeSummary(
("config/erp_semantics.php", "tests/Unit/FixTest.php"), 20, 2
)
workspace.run_args.side_effect = [
Mock(returncode=0, stdout="No syntax errors", stderr=""),
Mock(returncode=0, stdout="No syntax errors", stderr=""),
Mock(returncode=255, stdout="Parse error", stderr="unexpected token"),
Mock(returncode=0, stdout="No syntax errors", stderr=""),
Mock(returncode=0, stdout="No syntax errors", stderr=""),
]
workspace.diff.return_value = "--- app/Service.php\n+++ app/Service.php\n+application repair"
workspace.checkpoint.side_effect = ["checkpoint-1", "checkpoint-3"]
workspace.commit.return_value = "commit-id"
with patch("lai_automation.worker.GitWorkspace", return_value=workspace):
assert worker.run_once()
assert worker.agent.run.call_count == 3
workspace.restore_checkpoint.assert_called_once_with("checkpoint-1")
assert "last syntax-valid changes were restored" in worker.agent.run.call_args_list[2].args[1]
assert "application repair" in worker.agent.run.call_args_list[2].args[1]
assert worker._run_tests.call_count == 2
def test_failed_validation_gets_one_correction_before_pr() -> None:
report = FailureReport(
id=13,
chat_message_id=1,
source="automatic",
category="sql_execution",
user_feedback=None,
feedback_type=None,
expected_behavior=None,
prompt="List products",
response=None,
sql_query="SELECT missing_column FROM invoice_row",
error_message="Unknown column",
context={},
)
config = settings(mode="pr", dry_run=False, gitea_token="token")
worker = AutomationWorker(config)
worker.reports = Mock()
worker.reports.claim_next.return_value = report
worker.agent = Mock()
worker.agent.run.side_effect = ["first change", "corrected change"]
worker.gitea = Mock()
worker._run_tests = Mock(side_effect=[RuntimeError("Missing import"), []])
workspace = Mock()
workspace.path = Path("/tmp/nonexistent-workspace")
workspace.branch = "ai/failure-13"
workspace.summary.return_value = ChangeSummary(
("app/Services/Fix.php", "tests/Feature/FixTest.php"), 10, 1
)
workspace.run_args.return_value = Mock(returncode=0, stdout="No syntax errors", stderr="")
workspace.commit.return_value = "commit-id"
with patch("lai_automation.worker.GitWorkspace", return_value=workspace):
assert worker.run_once()
assert worker.agent.run.call_count == 2
assert "Missing import" in worker.agent.run.call_args_list[1].args[1]
assert worker._run_tests.call_count == 2
workspace.push_branch.assert_called_once()
worker.reports.complete.assert_called_once()
def test_no_change_response_gets_one_correction() -> None:
report = FailureReport(
id=14,
chat_message_id=1,
source="automatic",
category="sql_execution",
user_feedback=None,
feedback_type=None,
expected_behavior=None,
prompt="List products",
response=None,
sql_query=None,
error_message="Unknown column",
context={},
)
config = settings(mode="pr", dry_run=False, gitea_token="token")
worker = AutomationWorker(config)
worker.reports = Mock()
worker.reports.claim_next.return_value = report
worker.agent = Mock()
worker.agent.run.side_effect = ["no edit", "applied edit"]
worker.gitea = Mock()
worker._run_tests = Mock(return_value=[])
workspace = Mock()
workspace.path = Path("/tmp/nonexistent-workspace")
workspace.branch = "ai/failure-14"
workspace.summary.side_effect = [
ChangeSummary((), 0, 0),
ChangeSummary(("app/Services/Fix.php", "tests/Feature/FixTest.php"), 10, 1),
ChangeSummary(("app/Services/Fix.php", "tests/Feature/FixTest.php"), 10, 1),
]
workspace.run_args.return_value = Mock(returncode=0, stdout="No syntax errors", stderr="")
workspace.commit.return_value = "commit-id"
with patch("lai_automation.worker.GitWorkspace", return_value=workspace):
assert worker.run_once()
assert worker.agent.run.call_count == 2
assert "Agent produced no file changes" in worker.agent.run.call_args_list[1].args[1]
worker._run_tests.assert_called_once()
workspace.push_branch.assert_called_once()
def test_test_only_change_gets_correction_before_validation() -> None:
report = FailureReport(
id=49,
chat_message_id=1,
source="user_feedback",
category="incorrect_answer",
user_feedback="Token was not restored",
feedback_type="wrong_result",
expected_behavior="Return the real value",
prompt="Average price",
response="__LAI_UNKNOWN__",
sql_query="SELECT AVG(price)",
error_message=None,
context={},
)
config = settings(mode="pr", dry_run=False, gitea_token="token")
worker = AutomationWorker(config)
worker.reports = Mock()
worker.reports.claim_next.return_value = report
worker.agent = Mock()
worker.agent.run.side_effect = ["tests only", "production fix"]
worker.gitea = Mock()
worker._run_tests = Mock(return_value=[])
workspace = Mock()
workspace.path = Path("/tmp/nonexistent-workspace")
workspace.branch = "ai/failure-49"
workspace.summary.side_effect = [
ChangeSummary(("tests/Unit/FixTest.php",), 8, 0),
ChangeSummary(("app/Services/Fix.php", "tests/Unit/FixTest.php"), 15, 1),
ChangeSummary(("app/Services/Fix.php", "tests/Unit/FixTest.php"), 15, 1),
]
workspace.run_args.return_value = Mock(returncode=0, stdout="No syntax errors", stderr="")
workspace.commit.return_value = "commit-id"
with patch("lai_automation.worker.GitWorkspace", return_value=workspace):
assert worker.run_once()
assert worker.agent.run.call_count == 2
assert "changed only tests" in worker.agent.run.call_args_list[1].args[1]
worker._run_tests.assert_called_once()
workspace.cleanup.assert_called_once()