lai_automatisierung/tests/test_worker.py

import threading
from pathlib import Path
from unittest.mock import Mock, patch

from test_config import settings

from lai_automation.git import ChangeSummary
from lai_automation.models import FailureReport
from lai_automation.worker import AutomationWorker


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={},
    )
    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 test_database_credentials_are_only_passed_to_test_process() -> None:
    worker = AutomationWorker(settings())
    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_validation_commands_are_logged_without_output(caplog) -> None:
    worker = AutomationWorker(settings())
    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())
    workspace = Mock()
    workspace.run.return_value = Mock(
        returncode=1,
        stdout="DECISIVE ASSERTION\n" + "x" * 13_000,
        stderr="FINAL ERROR",
    )

    try:
        worker._run_tests(workspace)
    except RuntimeError as exception:
        failure = str(exception)
    else:
        raise AssertionError("Expected failed validation")

    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_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(("tests/Feature/FixTest.php",), 5, 0)
    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(("tests/Feature/FixTest.php",), 5, 0),
        ChangeSummary(("tests/Feature/FixTest.php",), 5, 0),
    ]
    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 "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()