from copy import deepcopy from types import SimpleNamespace import unittest from db.models import Assistant, ModelResource, TaskBackendResource, Tool from fastapi import FastAPI, HTTPException from httpx import ASGITransport, AsyncClient from routes import task_backends as routes from schemas import AssistantOut, AssistantUpsert, ToolUpsert from services.auth import require_admin from services.task_backend_resources import resolve_task_backend, validate_backend_binding, validate_backend_resource from task_backend_schemas import TaskBackendUpsert def opencode_body(**changes): return TaskBackendUpsert(name="远端任务", type="opencode", config={"url": "https://agent.test", "directory": "/work"}, **changes) class MemorySession: """Explicit test-only repository for API/config boundary tests.""" def __init__(self, *rows): self.rows = {(type(row), row.id): row for row in rows} async def get(self, model, key): return self.rows.get((model, key)) def add(self, row): self.rows[type(row), row.id] = row async def delete(self, row): self.rows.pop((type(row), row.id)) async def commit(self): pass async def refresh(self, row): pass async def execute(self, statement): model = statement.column_descriptions[0]["entity"] rows = [row for (kind, _), row in self.rows.items() if kind is model] return SimpleNamespace(scalars=lambda: SimpleNamespace(all=lambda: rows)) class TaskBackendSchemaTests(unittest.TestCase): def test_task_is_not_an_ordinary_tool_resource(self): with self.assertRaises(ValueError): ToolUpsert(name="旧工具", functionName="coding_task", definition={"type": "task", "config": {"url": "https://agent.test", "directory": "/work"}}) def test_backend_config_requires_real_supported_fields(self): with self.assertRaisesRegex(ValueError, "Extra inputs"): TaskBackendUpsert(name="后端", type="opencode", config={"url": "https://agent.test", "directory": "/work", "grantAllPermissions": True}) with self.assertRaisesRegex(ValueError, "同时填写"): TaskBackendUpsert(name="后端", type="opencode", config={"url": "https://agent.test", "directory": "/work", "modelId": "model"}) def test_reasoner_rejects_duplicate_tools_and_skills(self): for extra in [ {"toolBindings": [{"toolId": "http_1"}, {"toolId": "http_1"}]}, {"skills": [{"name": "review", "content": "a"}, {"name": " review ", "content": "b"}]}, ]: with self.subTest(extra=extra), self.assertRaisesRegex(ValueError, "重复"): TaskBackendUpsert(name="推理", type="reasoner", config={"modelResourceId": "llm", **extra}) def test_assistant_binding_preserves_all_instruction_sources(self): for source in [ {"source": "model"}, {"source": "literal", "value": "固定任务"}, {"source": "variable", "variableName": "goal"}, {"source": "template", "value": "处理 {{goal}}"}, ]: with self.subTest(source=source): body = AssistantUpsert(name="助手", taskBackend={"backendId": "legacy_tool_id", "instructions": "补充说明", "instructionSource": source}) out = AssistantOut(id="asst", **body.model_dump()).model_dump(by_alias=True) self.assertEqual(out["taskBackend"]["instructionSource"], source) self.assertEqual(out["taskBackend"]["instructions"], "补充说明") def test_task_backend_is_prompt_pipeline_only(self): for options in [{"type": "workflow"}, {"runtimeMode": "realtime"}, {"type": "dify"}]: with self.subTest(options=options), self.assertRaisesRegex(ValueError, "提示词 Pipeline"): AssistantUpsert(name="助手", taskBackend={"backendId": "backend_1"}, **options) class TaskBackendValidationTests(unittest.IsolatedAsyncioTestCase): def setUp(self): self.model = ModelResource(id="llm", name="深思考", capability="LLM", interface_type="openai-llm", values={}, secrets={"apiKey": "model-secret"}, enabled=True) self.session = MemorySession(self.model) async def test_reasoning_strength_requires_explicit_model_capability(self): body = TaskBackendUpsert(name="推理", type="reasoner", config={"modelResourceId": "llm", "reasoningEffort": "high"}) with self.assertRaisesRegex(ValueError, "未声明"): await validate_backend_resource(self.session, body) self.model.values = {"supportsReasoningEffort": True} await validate_backend_resource(self.session, body) async def test_reasoner_rejects_client_tools_and_unknown_required_inputs(self): tool = Tool(id="tool", name="工具", function_name="read_data", type="client", status="active", definition={"config": {}}) self.session.add(tool) body = TaskBackendUpsert(name="推理", type="reasoner", config={"modelResourceId": "llm", "toolBindings": [{"toolId": "tool"}]}) with self.assertRaisesRegex(ValueError, "HTTP/MCP"): await validate_backend_resource(self.session, body) tool.type = "http" tool.definition = {"config": {"parameters": [{"name": "query", "required": True, "type": "string"}]}} with self.assertRaisesRegex(ValueError, "缺少必填"): await validate_backend_resource(self.session, body) async def test_template_sources_are_checked_against_actual_assistant(self): resource = TaskBackendResource(id="backend", **routes._payload(opencode_body())) self.session.add(resource) body = AssistantUpsert(name="助手", taskBackend={"backendId": "backend", "instructionSource": {"source": "template", "value": "处理 {{missing}}"}}) with self.assertRaisesRegex(ValueError, "未声明变量"): await validate_backend_binding(self.session, body) body.dynamic_variable_definitions = {"missing": {"type": "string"}} await validate_backend_binding(self.session, body) async def test_runtime_snapshot_resolves_backend_and_model_separately(self): resource = TaskBackendResource(id="backend", name="推理", type="reasoner", config={"model_resource_id": "llm", "reasoning_effort": "provider_default"}, secrets={}, enabled=True) self.session.add(resource) assistant = Assistant(id="assistant", name="助手", type="prompt", runtime_mode="pipeline", task_backend={"backend_id": "backend", "instruction_source": {"source": "literal", "value": "固定任务"}}, dynamic_variable_definitions={}) resolved = await resolve_task_backend(self.session, assistant) self.assertEqual(resolved.model.id, "llm") self.assertEqual(resolved.model.secrets["apiKey"], "model-secret") self.assertEqual(resolved.instruction_source["value"], "固定任务") self.assertEqual(resolved.config["max_concurrency"], 1) async def test_masked_secret_preserves_stored_value_and_type_switch_discards_it(self): stored = TaskBackendResource(type="opencode", secrets={"password": "stored-secret"}) payload = routes._payload(opencode_body(secrets={"password": "****"}), stored) self.assertEqual(payload["secrets"]["password"], "stored-secret") reasoner = TaskBackendUpsert(name="推理", type="reasoner", config={"modelResourceId": "llm"}) self.assertEqual(routes._payload(reasoner, stored)["secrets"], {}) async def test_resource_api_requires_admin_and_masks_passwords(self): app = FastAPI() app.include_router(routes.router) async def session_dependency(): yield self.session app.dependency_overrides[routes.get_session] = session_dependency async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: self.assertEqual((await client.get("/api/task-backends")).status_code, 401) app.dependency_overrides[require_admin] = lambda: True response = await client.post("/api/task-backends", json=opencode_body(secrets={"password": "keep-me-private"}).model_dump(by_alias=True)) self.assertEqual(response.status_code, 200, response.text) backend_id = response.json()["id"] self.assertNotIn("keep-me-private", response.text) data = response.json() data["name"] = "修改后端" response = await client.put(f"/api/task-backends/{backend_id}", json=data) self.assertEqual(response.status_code, 200, response.text) stored = await self.session.get(TaskBackendResource, backend_id) self.assertEqual(stored.secrets["password"], "keep-me-private") duplicate = await client.post(f"/api/task-backends/{backend_id}/duplicate") self.assertEqual(duplicate.status_code, 200, duplicate.text) self.assertNotIn("keep-me-private", duplicate.text) self.assertNotEqual(duplicate.json()["id"], backend_id) self.assertEqual((await client.delete(f"/api/task-backends/{backend_id}")).status_code, 200) if __name__ == "__main__": unittest.main()