from unittest.mock import patch 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, ProviderConnectionError, ) from .config import AgentRuntimeConfig from .gateway import AgentExecutionResult, StubAgentRunnerGateway from .models import ApprovalStatus, ModelProviderConfig, RunStatus, ToolCallStatus from .orchestration import execute_run from .providers import get_provider from .secrets import decrypt_api_key, encrypt_api_key from .services import ( ProviderVerificationResult, create_run, disconnect_provider, finish_tool_call, request_approval, resolve_approval, resolve_runtime_model_config, save_verified_provider_config, start_tool_call, transition_run, verify_provider_api_key, ) 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") class FakeModelsResponse: """模拟只返回公开模型标识的服务商模型目录。""" def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): return False def read(self): return b'{"data":[{"id":"model-a"},{"id":"model-b"}]}' class ProviderVerificationTests(SimpleTestCase): """验证连接测试只访问模型目录，且凭据不会进入 URL。""" def test_openai_compatible_verification_uses_bearer_header(self): captured = {} def opener(request, timeout): captured["url"] = request.full_url captured["authorization"] = request.get_header("Authorization") captured["timeout"] = timeout return FakeModelsResponse() result = verify_provider_api_key(get_provider("deepseek"), "safe-test-key", opener=opener) self.assertEqual(result.models, ("model-a", "model-b")) self.assertEqual(captured["authorization"], "Bearer safe-test-key") self.assertNotIn("safe-test-key", captured["url"]) self.assertEqual(captured["timeout"], 12) def test_provider_requiring_extra_configuration_is_rejected_without_request(self): with self.assertRaises(ProviderConnectionError) as context: verify_provider_api_key(get_provider("azure_openai"), "safe-test-key") self.assertEqual(context.exception.code, "extra_configuration_required") @override_settings(MODEL_API_KEY_ENCRYPTION_KEY="test-only-model-key-32-characters-minimum") class ModelProviderConfigTests(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") self.provider = get_provider("openai") self.result = ProviderVerificationResult(("model-a", "model-b")) def test_api_key_round_trip_uses_ciphertext(self): ciphertext = encrypt_api_key("safe-test-key") self.assertNotIn("safe-test-key", ciphertext) self.assertEqual(decrypt_api_key(ciphertext), "safe-test-key") def test_verified_config_is_user_owned_and_resolves_for_runtime(self): config = save_verified_provider_config( self.alice, self.provider, "safe-test-key", self.result, model_id="model-a", ) self.assertTrue(config.is_default) runtime = resolve_runtime_model_config(self.alice) self.assertEqual(runtime.api_key, "safe-test-key") self.assertEqual(runtime.model, "model-a") user_config = AgentRuntimeConfig.from_user(self.alice) self.assertEqual(user_config.provider_code, "openai") self.assertEqual(user_config.protocol, "openai_responses") self.assertEqual(user_config.base_url, "https://api.openai.com/v1") with self.assertRaises(AgentConfigurationError): resolve_runtime_model_config(self.bob) def test_disconnect_promotes_remaining_provider(self): first = save_verified_provider_config( self.alice, self.provider, "safe-test-key", self.result, model_id="model-a" ) second = save_verified_provider_config( self.alice, get_provider("deepseek"), "another-safe-key", self.result, model_id="model-b", ) self.assertTrue(first.is_default) self.assertFalse(second.is_default) disconnect_provider(self.alice, "openai") second.refresh_from_db() self.assertTrue(second.is_default) def test_provider_page_requires_login_and_lists_catalog(self): response = self.client.get(reverse("agent_runtime:provider-list")) self.assertEqual(response.status_code, 302) self.client.force_login(self.alice) response = self.client.get(reverse("agent_runtime:provider-list")) self.assertContains(response, "阿里云百炼") self.assertContains(response, "小米 MiMo") self.assertContains(response, "共 23 个内置服务商") @patch("agent_runtime.views.verify_provider_api_key") def test_configure_page_saves_verified_key_without_rendering_it(self, verify): verify.return_value = self.result self.client.force_login(self.alice) response = self.client.post( reverse("agent_runtime:provider-configure", args=("openai",)), {"api_key": "safe-test-key", "model_id": "model-a", "use_as_default": "on"}, ) self.assertRedirects(response, reverse("agent_runtime:provider-list")) config = ModelProviderConfig.objects.get(owner=self.alice, provider_code="openai") self.assertNotIn("safe-test-key", config.encrypted_api_key) page = self.client.get(f"{reverse('agent_runtime:provider-list')}?provider=openai") self.assertNotContains(page, "safe-test-key") self.assertContains(page, config.key_hint) @patch("agent_runtime.views.verify_provider_api_key") def test_failed_reconfiguration_does_not_replace_existing_key(self, verify): config = save_verified_provider_config( self.alice, self.provider, "original-safe-key", self.result, model_id="model-a" ) original_ciphertext = config.encrypted_api_key verify.side_effect = ProviderConnectionError("invalid_api_key", "API Key 无效。") self.client.force_login(self.alice) response = self.client.post( reverse("agent_runtime:provider-configure", args=("openai",)), {"api_key": "incorrect-key", "model_id": "model-b"}, ) self.assertEqual(response.status_code, 400) config.refresh_from_db() self.assertEqual(config.encrypted_api_key, original_ciphertext) self.assertEqual(decrypt_api_key(config.encrypted_api_key), "original-safe-key") def test_connected_filter_never_exposes_another_users_provider(self): save_verified_provider_config( self.bob, self.provider, "bob-safe-key", self.result, model_id="model-a" ) self.client.force_login(self.alice) response = self.client.get( reverse("agent_runtime:provider-list"), {"category": "connected"} ) self.assertNotContains(response, "OpenAI")