43 lines
2.0 KiB
Python
43 lines
2.0 KiB
Python
from django.contrib.auth import get_user_model
|
|
from django.test import TestCase
|
|
from django.urls import reverse
|
|
|
|
from common.exceptions import InvalidStateTransition
|
|
|
|
from .models import RunStatus
|
|
from .services import create_run, finish_tool_call, start_tool_call, transition_run
|
|
|
|
|
|
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_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})
|