steven_li
ebfa242862
添加完整的Outlook MCP集成，包括邮件和日历功能，通过AuthZ模式进行认证和权限管理， 支持邮箱连接、断开、状态检查和数据同步等功能。 fix(config): 统一配置文件路径从.nanobot到.beaver 将配置文件路径从/root/.nanobot统一更改为/root/.beaver，更新Dockerfile中的环境变量定义， 确保所有组件使用一致的配置目录结构。 feat(agent): 添加代理删除功能和助手身份提示 为代理注册表添加delete_agent方法，实现代理的动态删除功能；同时添加海狸助手身份提示， 确保AI助手在交互中保持一致的身份认知。 feat(engine): 增强引擎循环并添加意图决策快照 扩展AgentLoop类，添加intent_agent_decision参数用于意图驱动的代理决策，并在会话中记录 决策快照，便于后续分析和调试。 feat(authz): 扩展认证客户端功能 为AuthzClient添加设置权限、用户注册、后端注册和Outlook设置管理等新方法，增强系统 的认证和授权能力。
71 lines
2.9 KiB
Python
71 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from beaver.interfaces.web.app import create_app
|
|
from beaver.services.agent_service import AgentService
|
|
|
|
|
|
def test_workspace_browser_api_manages_workspace_files(tmp_path: Path) -> None:
|
|
service = AgentService(workspace=tmp_path)
|
|
app = create_app(service=service, manage_service_lifecycle=False)
|
|
|
|
with TestClient(app) as client:
|
|
root = client.get("/api/workspace/browse")
|
|
mkdir = client.post("/api/workspace/mkdir", params={"path": "docs"})
|
|
upload = client.post(
|
|
"/api/workspace/upload",
|
|
data={"path": "docs"},
|
|
files={"file": ("hello.txt", b"hello workspace", "text/plain")},
|
|
)
|
|
docs = client.get("/api/workspace/browse", params={"path": "docs"})
|
|
download = client.get("/api/workspace/download", params={"path": "docs/hello.txt"})
|
|
deleted = client.delete("/api/workspace/delete", params={"path": "docs/hello.txt"})
|
|
after_delete = client.get("/api/workspace/browse", params={"path": "docs"})
|
|
|
|
assert root.status_code == 200
|
|
assert root.json()["path"] == ""
|
|
assert all(item["name"] != "docs" for item in root.json()["items"])
|
|
assert mkdir.status_code == 200
|
|
assert mkdir.json()["path"] == "docs"
|
|
assert upload.status_code == 200
|
|
assert upload.json()["path"] == "docs/hello.txt"
|
|
assert docs.status_code == 200
|
|
assert [item["name"] for item in docs.json()["items"]] == ["hello.txt"]
|
|
assert download.status_code == 200
|
|
assert download.content == b"hello workspace"
|
|
assert deleted.status_code == 200
|
|
assert deleted.json() == {"ok": True}
|
|
assert after_delete.status_code == 200
|
|
assert after_delete.json()["items"] == []
|
|
|
|
|
|
def test_attachment_file_api_round_trips_uploaded_file(tmp_path: Path) -> None:
|
|
service = AgentService(workspace=tmp_path)
|
|
app = create_app(service=service, manage_service_lifecycle=False)
|
|
|
|
with TestClient(app) as client:
|
|
upload = client.post(
|
|
"/api/files/upload",
|
|
data={"session_id": "web:test"},
|
|
files={"file": ("note.txt", b"hello attachment", "text/plain")},
|
|
)
|
|
file_id = upload.json()["file_id"]
|
|
listed = client.get("/api/files", params={"session_id": "web:test"})
|
|
download = client.get(f"/api/files/{file_id}")
|
|
deleted = client.delete(f"/api/files/{file_id}")
|
|
missing = client.get(f"/api/files/{file_id}")
|
|
|
|
assert upload.status_code == 200
|
|
assert upload.json()["name"] == "note.txt"
|
|
assert upload.json()["url"] == f"/api/files/{file_id}"
|
|
assert listed.status_code == 200
|
|
assert [item["file_id"] for item in listed.json()] == [file_id]
|
|
assert download.status_code == 200
|
|
assert download.content == b"hello attachment"
|
|
assert deleted.status_code == 200
|
|
assert deleted.json() == {"ok": True}
|
|
assert missing.status_code == 404
|