from django.contrib.auth import get_user_model from django.db import IntegrityError, transaction from django.test import SimpleTestCase, TestCase, override_settings from django.urls import reverse from common.exceptions import AgentConfigurationError, InvalidStateTransition, PermissionDenied from .config import AgentRuntimeConfig from .gateway import AgentExecutionResult, StubAgentRunnerGateway from .models import ApprovalStatus, RunStatus, ToolCallStatus from .orchestration import execute_run from .services import ( create_run, finish_tool_call, request_approval, resolve_approval, start_tool_call, transition_run, ) class ExplodingGateway: """模拟外部边界抛错，并故意在异常文本中携带不应落库的秘密。""" def run(self, request): raise RuntimeError("api_key=should-not-be-stored") class AgentRunTests(TestCase): """覆盖运行状态、事件顺序、脱敏和跨用户拒绝路径。""" def setUp(self): users = get_user_model().objects self.alice = users.create_user(username="alice", password="safe-pass-123") self.bob = users.create_user(username="bob", password="safe-pass-123") def test_run_lifecycle_and_sensitive_summary(self): run = create_run(self.alice, "基线运行", {"api_key": "secret", "query": "Python"}) self.assertEqual(run.input_summary["api_key"], "***") transition_run(run.id, RunStatus.RUNNING) finished = transition_run(run.id, RunStatus.SUCCEEDED, output={"count": 1}) self.assertEqual(finished.status, RunStatus.SUCCEEDED) self.assertEqual(list(finished.events.values_list("sequence", flat=True)), [1, 2, 3]) with self.assertRaises(InvalidStateTransition): transition_run(run.id, RunStatus.RUNNING) def test_cross_user_detail_returns_404(self): run = create_run(self.alice, "私有运行") self.client.force_login(self.bob) response = self.client.get(reverse("agent_runtime:run-detail", args=(run.id,))) self.assertEqual(response.status_code, 404) def test_run_list_only_contains_current_user_data(self): create_run(self.alice, "爱丽丝的运行") create_run(self.bob, "鲍勃的运行") self.client.force_login(self.alice) response = self.client.get(reverse("agent_runtime:run-list")) self.assertEqual(response.status_code, 200) self.assertContains(response, "爱丽丝的运行") self.assertNotContains(response, "鲍勃的运行") def test_tool_call_is_recorded_and_sanitized(self): run = create_run(self.alice, "工具运行") transition_run(run.id, RunStatus.RUNNING) call = start_tool_call(run.id, "call-1", "demo_tool", {"password": "secret"}) self.assertEqual(call.arguments_summary["password"], "***") finished = finish_tool_call(call.id, result={"count": 1}) self.assertEqual(finished.status, "succeeded") self.assertEqual(finished.result_summary, {"count": 1}) def test_tool_call_failure_and_duplicate_finish_are_recorded(self): run = create_run(self.alice, "失败工具运行") transition_run(run.id, RunStatus.RUNNING) call = start_tool_call(run.id, "call-1", "demo_tool", idempotency_key="same-operation") failed = finish_tool_call(call.id, error_code="timeout", error_summary="请求超时") self.assertEqual(failed.status, ToolCallStatus.FAILED) self.assertEqual(failed.error_code, "timeout") with self.assertRaises(InvalidStateTransition): finish_tool_call(call.id, result={"unexpected": True}) def test_duplicate_tool_idempotency_key_is_rejected(self): run = create_run(self.alice, "幂等工具运行") transition_run(run.id, RunStatus.RUNNING) start_tool_call(run.id, "call-1", "demo_tool", idempotency_key="same-operation") with self.assertRaises(IntegrityError), transaction.atomic(): start_tool_call(run.id, "call-2", "demo_tool", idempotency_key="same-operation") def test_approval_can_be_approved_once_by_owner(self): run = create_run(self.alice, "确认运行") transition_run(run.id, RunStatus.RUNNING) approval = request_approval(run.id, "approval-1", "external_action", {"token": "secret"}) self.assertEqual(approval.request_summary["token"], "***") resolved = resolve_approval(self.alice, approval.id, True, {"reason": "允许"}) self.assertEqual(resolved.status, ApprovalStatus.APPROVED) run.refresh_from_db() self.assertEqual(run.status, RunStatus.RUNNING) with self.assertRaises(InvalidStateTransition): resolve_approval(self.alice, approval.id, True) def test_approval_rejects_cross_user_and_can_cancel_run(self): run = create_run(self.alice, "拒绝确认运行") transition_run(run.id, RunStatus.RUNNING) approval = request_approval(run.id, "approval-1", "external_action") with self.assertRaises(PermissionDenied): resolve_approval(self.bob, approval.id, False) resolve_approval(self.alice, approval.id, False) run.refresh_from_db() self.assertEqual(run.status, RunStatus.CANCELLED) def test_stub_gateway_drives_successful_persistent_run(self): run = create_run(self.alice, "Stub 成功运行", {"query": "Python"}) gateway = StubAgentRunnerGateway(AgentExecutionResult(True, {"count": 2})) finished = execute_run(run.id, gateway) self.assertEqual(finished.status, RunStatus.SUCCEEDED) self.assertEqual(finished.output_summary, {"count": 2}) self.assertEqual(finished.events.count(), 3) def test_stub_gateway_failure_and_exception_reach_failed_state(self): failed_run = create_run(self.alice, "Stub 失败运行") failed_gateway = StubAgentRunnerGateway( AgentExecutionResult(False, error_code="model_error", error_summary="模型失败") ) failed = execute_run(failed_run.id, failed_gateway) self.assertEqual(failed.status, RunStatus.FAILED) self.assertEqual(failed.error_code, "model_error") exploding_run = create_run(self.alice, "Stub 异常运行") exploded = execute_run(exploding_run.id, ExplodingGateway()) self.assertEqual(exploded.status, RunStatus.FAILED) self.assertNotIn("should-not-be-stored", exploded.error_summary) class AgentRuntimeConfigTests(SimpleTestCase): """验证真实执行配置只在调用边界检查，不影响其他 Django 功能。""" @override_settings(OPENAI_API_KEY="", OPENAI_MODEL="", AGENT_DEFAULT_NAME="job_research") def test_real_execution_requires_key_and_model(self): with self.assertRaises(AgentConfigurationError): AgentRuntimeConfig.from_settings().require_real_execution() @override_settings( OPENAI_API_KEY="test-key", OPENAI_MODEL="test-model", AGENT_DEFAULT_NAME="job_research" ) def test_complete_configuration_is_accepted_without_external_call(self): config = AgentRuntimeConfig.from_settings().require_real_execution() self.assertEqual(config.model, "test-model")