from django.contrib.auth import get_user_model from django.core.management import call_command from apps.agents.models import AgentDefinition, AgentRun from apps.agents.services import build_config_snapshot from apps.modelhub.services import CompletionResult from .helpers import create_agent, create_deployment def login_admin(client): call_command("bootstrap_governance") user = get_user_model().objects.get(username="admin") client.force_login(user) return user def test_admin_publishes_and_runs_agent(client, db, monkeypatch): login_admin(client) definition, version = create_agent() outputs = iter( [ CompletionResult('{"summary":"ok"}', 3, 2), CompletionResult('{"reply":"ok"}', 3, 2), ] ) monkeypatch.setattr( "apps.modelhub.services._request_completion", lambda *args, **kwargs: next(outputs), ) published = client.post(f"/api/agents/{definition.id}/publish") assert published.status_code == 200 assert published.json()["status"] == "published" run = client.post( f"/api/agents/{definition.id}/runs", data={"version": 1, "input": {"problem": "无法登录"}}, content_type="application/json", ) assert run.status_code == 200 assert run.json()["status"] == "succeeded" public_id = run.json()["public_id"] detail = client.get(f"/api/agent-runs/{public_id}") assert detail.status_code == 200 assert detail.json()["config_snapshot"]["agent"]["version"] == 1 assert all( step["model_deployment_id"] == version.model_deployment_id for step in detail.json()["steps"] ) def test_publishing_rejects_draft_skill(client, db): login_admin(client) definition, version = create_agent() binding = version.steps.first() binding.skill_version.status = "draft" binding.skill_version.save(update_fields=["status"]) response = client.post(f"/api/agents/{definition.id}/publish") assert response.status_code == 400 assert "已发布 Skill" in response.json()["detail"] def test_publish_returns_structured_error_code_for_unverified_model(client, db): login_admin(client) definition, version = create_agent() version.model_deployment.structured_output_test_status = "untested" version.model_deployment.save(update_fields=["structured_output_test_status"]) response = client.post(f"/api/agents/{definition.id}/publish") assert response.status_code == 400 assert response.json()["detail"].startswith("structured_output_not_verified:") def test_draft_steps_can_override_model_and_clone_binding(client, db): login_admin(client) definition, version = create_agent() override = create_deployment( capability="vision_chat", provider=version.model_deployment.provider, suffix="-vision", ) steps = [ { "code": step.code, "skill_version_id": step.skill_version_id, "model_deployment_id": override.id if index == 0 else None, "input_mapping": step.input_mapping, } for index, step in enumerate(version.steps.all()) ] response = client.patch( f"/api/agents/{definition.id}/draft", data={"steps": steps}, content_type="application/json", ) assert response.status_code == 200 response_steps = response.json()["steps"] assert response_steps[0]["model_deployment_id"] == override.id assert response_steps[0]["model_display_name"] == override.display_name assert response_steps[0]["model_capability"] == "vision_chat" assert response_steps[1]["model_deployment_id"] is None assert response_steps[1]["model_display_name"] == version.model_deployment.display_name assert client.post(f"/api/agents/{definition.id}/publish").status_code == 200 clone = client.post(f"/api/agents/{definition.id}/draft") assert clone.status_code == 200 assert clone.json()["steps"][0]["model_deployment_id"] == override.id assert clone.json()["steps"][1]["model_deployment_id"] is None def test_publishing_rejects_non_chat_step_model(client, db): login_admin(client) definition, version = create_agent() embedding = create_deployment( capability="embedding", provider=version.model_deployment.provider, suffix="-embedding", ) binding = version.steps.first() binding.model_deployment = embedding binding.save(update_fields=["model_deployment"]) response = client.post(f"/api/agents/{definition.id}/publish") assert response.status_code == 400 assert "必须绑定文本或视觉模型" in response.json()["detail"] def test_agent_run_permission_is_required(client, db): definition, _ = create_agent(published=True) user = get_user_model().objects.create_user(username="requester", password="password") client.force_login(user) response = client.post( f"/api/agents/{definition.id}/runs", data={"version": 1, "input": {"problem": "无法登录"}}, content_type="application/json", ) assert response.status_code == 403 def test_agent_run_creator_can_view_own_detail_without_global_run_permission(client, db): definition, version = create_agent(published=True) owner = get_user_model().objects.create_user(username="run-owner") outsider = get_user_model().objects.create_user(username="run-outsider") run = AgentRun.objects.create( definition=definition, agent_version=version, input_payload={"problem": "无法登录"}, config_snapshot=build_config_snapshot(version), created_by=owner, total_steps=version.steps.count(), ) client.force_login(owner) owner_response = client.get(f"/api/agent-runs/{run.public_id}") client.force_login(outsider) outsider_response = client.get(f"/api/agent-runs/{run.public_id}") assert owner_response.status_code == 200 assert outsider_response.status_code == 403 def test_agent_run_list_is_paginated_after_filtering(client, db): login_admin(client) definition, version = create_agent(published=True) runs = [ AgentRun.objects.create( definition=definition, agent_version=version, status="failed" if index < 2 else "queued", input_payload={"sequence": index}, ) for index in range(25) ] response = client.get("/api/agent-runs?status=queued&page=2&page_size=10") assert response.status_code == 200 payload = response.json() assert payload["page"] == 2 assert payload["page_size"] == 10 assert payload["total"] == 23 assert payload["total_pages"] == 3 assert len(payload["items"]) == 10 assert all(item["status"] == "queued" for item in payload["items"]) assert payload["items"][0]["public_id"] == str(runs[14].public_id) def test_agent_run_list_clamps_invalid_pagination_values(client, db): login_admin(client) definition, version = create_agent(published=True) AgentRun.objects.create( definition=definition, agent_version=version, input_payload={"problem": "无法登录"}, ) response = client.get("/api/agent-runs?page=0&page_size=500") assert response.status_code == 200 assert response.json()["page"] == 1 assert response.json()["page_size"] == 100 def test_queued_run_can_be_cancelled(client, db, monkeypatch): login_admin(client) definition, version = create_agent(published=True) run = AgentRun.objects.create( definition=definition, agent_version=version, input_payload={"problem": "无法登录"}, config_snapshot=build_config_snapshot(version), total_steps=version.steps.count(), ) revoked = [] monkeypatch.setattr( "apps.agents.api.AsyncResult.revoke", lambda self, **kwargs: revoked.append(kwargs), ) run.task_id = "queued-task-id" run.save(update_fields=["task_id"]) response = client.post(f"/api/agent-runs/{run.public_id}/cancel") assert response.status_code == 200 assert response.json()["status"] == "cancelled" assert response.json()["cancellation_requested_at"] is not None assert revoked == [{"terminate": False}] def test_seed_command_creates_support_agent(client, db): deployment = create_deployment() business_agent = AgentDefinition.objects.create( code="support-analysis", name="支持分析 Agent", ) call_command("seed_ai_demo", deployment_id=deployment.id) demo_agent = AgentDefinition.objects.get(code="support-analysis-demo") assert demo_agent.versions.get(status="draft").steps.count() == 3 business_agent.refresh_from_db() assert business_agent.versions.count() == 0 assert business_agent.active_version is None