57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
from unittest.mock import Mock
|
|
|
|
from test_config import settings
|
|
|
|
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["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"
|