|
|
@@ -0,0 +1,138 @@
|
|
|
+"""
|
|
|
+tests/test_workspace_provision.py — 创建智能体实例时自动开辟工作目录。
|
|
|
+
|
|
|
+布局参照 Workspace/multi-reviewer67：
|
|
|
+ <base>/<名>/ work_dir（最终交付物 + Bash CWD）
|
|
|
+ <base>/<名>/data source_dir
|
|
|
+ <base>/<名>/workspace run_dir（bare 模式，run_* 直接建在其下）
|
|
|
+"""
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import json
|
|
|
+import os
|
|
|
+
|
|
|
+import pytest
|
|
|
+
|
|
|
+os.environ.setdefault("AGENTPAAS_DATABASE_URL", "sqlite:///:memory:")
|
|
|
+os.environ.setdefault("AGENTPAAS_TESTING", "1")
|
|
|
+
|
|
|
+
|
|
|
+# ── 单元：provision 函数 ─────────────────────────────────────────────────────
|
|
|
+
|
|
|
+def test_provision_layout(tmp_path):
|
|
|
+ from agentpaas.engine.workspace_provision import provision_agent_workspace
|
|
|
+
|
|
|
+ d = provision_agent_workspace("审稿助手", base=str(tmp_path))
|
|
|
+ root = os.path.join(str(tmp_path), "审稿助手")
|
|
|
+ assert d["work_dir"] == os.path.abspath(root)
|
|
|
+ assert d["source_dir"] == os.path.join(os.path.abspath(root), "data")
|
|
|
+ assert d["run_dir"] == os.path.join(os.path.abspath(root), "workspace")
|
|
|
+ assert os.path.isdir(d["source_dir"]) and os.path.isdir(d["run_dir"])
|
|
|
+
|
|
|
+
|
|
|
+def test_provision_duplicate_names_get_suffix(tmp_path):
|
|
|
+ from agentpaas.engine.workspace_provision import provision_agent_workspace
|
|
|
+
|
|
|
+ d1 = provision_agent_workspace("助手", base=str(tmp_path))
|
|
|
+ d2 = provision_agent_workspace("助手", base=str(tmp_path))
|
|
|
+ d3 = provision_agent_workspace("助手", base=str(tmp_path))
|
|
|
+ assert d1["work_dir"].endswith("助手")
|
|
|
+ assert d2["work_dir"].endswith("助手-2")
|
|
|
+ assert d3["work_dir"].endswith("助手-3")
|
|
|
+
|
|
|
+
|
|
|
+def test_provision_slug_sanitises_path_chars(tmp_path):
|
|
|
+ from agentpaas.engine.workspace_provision import provision_agent_workspace
|
|
|
+
|
|
|
+ d = provision_agent_workspace("../evil/name v2", base=str(tmp_path))
|
|
|
+ # 不逃逸 base；路径分隔符被清洗
|
|
|
+ assert os.path.commonpath([d["work_dir"], str(tmp_path)]) == str(
|
|
|
+ os.path.abspath(str(tmp_path)))
|
|
|
+ assert "/evil/" not in d["work_dir"].replace(str(tmp_path), "")
|
|
|
+
|
|
|
+
|
|
|
+def test_provision_failure_returns_empty(monkeypatch):
|
|
|
+ from agentpaas.engine import workspace_provision as wp
|
|
|
+
|
|
|
+ monkeypatch.setattr(wp.os, "makedirs",
|
|
|
+ lambda *a, **k: (_ for _ in ()).throw(OSError("ro")))
|
|
|
+ d = wp.provision_agent_workspace("x", base="/nonexistent-base")
|
|
|
+ assert d == {"work_dir": "", "source_dir": "", "run_dir": ""}
|
|
|
+
|
|
|
+
|
|
|
+# ── 集成：两个创建端点 ───────────────────────────────────────────────────────
|
|
|
+
|
|
|
+@pytest.fixture()
|
|
|
+def api_client(tmp_path, monkeypatch):
|
|
|
+ """TestClient + 内存 DB + 租户/key + workspace_base 指到 tmp。"""
|
|
|
+ import secrets
|
|
|
+ from fastapi.testclient import TestClient
|
|
|
+ from agentpaas.api.app import app
|
|
|
+ from agentpaas.api.middleware.auth import hash_key
|
|
|
+ from agentpaas.config import settings
|
|
|
+ from agentpaas.db.models import Database, gen_id, now_utc
|
|
|
+ import agentpaas.db.session as _session_mod
|
|
|
+
|
|
|
+ ws_base = tmp_path / "Workspace"
|
|
|
+ monkeypatch.setattr(settings, "workspace_base", str(ws_base))
|
|
|
+
|
|
|
+ prev = _session_mod._db
|
|
|
+ _session_mod._db = Database("sqlite:///:memory:")
|
|
|
+ db = _session_mod._db
|
|
|
+ tid, uid = gen_id("tn_"), gen_id("usr_")
|
|
|
+ raw_key = f"ap_{secrets.token_hex(16)}"
|
|
|
+ now = now_utc()
|
|
|
+ db.execute("INSERT INTO tenants (id, name, plan, status, created_at) "
|
|
|
+ "VALUES (?, 'test', 'free', 'active', ?)", (tid, now))
|
|
|
+ db.execute("INSERT INTO users (id, tenant_id, email, role, created_at) "
|
|
|
+ "VALUES (?, ?, '', 'admin', ?)", (uid, tid, now))
|
|
|
+ db.execute(
|
|
|
+ "INSERT INTO api_keys (id, tenant_id, user_id, key_hash, key_prefix, name, "
|
|
|
+ "scopes, rate_limit, status, created_at) "
|
|
|
+ "VALUES (?, ?, ?, ?, ?, 'test', ?, 600, 'active', ?)",
|
|
|
+ (gen_id("key_"), tid, uid, hash_key(raw_key), raw_key[:8],
|
|
|
+ json.dumps(["agents:*"]), now))
|
|
|
+ db.commit()
|
|
|
+ with TestClient(app) as client:
|
|
|
+ yield client, raw_key, str(ws_base), db
|
|
|
+ _session_mod._db = prev
|
|
|
+
|
|
|
+
|
|
|
+def _auth(k):
|
|
|
+ return {"Authorization": f"Bearer {k}"}
|
|
|
+
|
|
|
+
|
|
|
+_MIN_CONFIG = {"name": "目录测试", "type": "simple",
|
|
|
+ "model": {"name": "ollama/qwen2.5:7b"}, "systemPrompt": "t"}
|
|
|
+
|
|
|
+
|
|
|
+def test_create_agent_auto_provisions_dirs(api_client):
|
|
|
+ client, key, ws_base, db = api_client
|
|
|
+ r = client.post("/api/v1/agents", headers=_auth(key),
|
|
|
+ json={"name": "目录测试", "config": _MIN_CONFIG})
|
|
|
+ assert r.status_code in (200, 201), r.text[:300]
|
|
|
+ body = r.json()
|
|
|
+ assert body["work_dir"].endswith("目录测试")
|
|
|
+ assert body["run_dir"] == os.path.join(body["work_dir"], "workspace")
|
|
|
+ assert os.path.isdir(body["source_dir"])
|
|
|
+
|
|
|
+ # DB 行一致
|
|
|
+ row = db.fetchone("SELECT work_dir, source_dir, run_dir FROM agents WHERE id=?",
|
|
|
+ (body["agent_id"],))
|
|
|
+ assert row["work_dir"] == body["work_dir"]
|
|
|
+ assert row["run_dir"] == body["run_dir"]
|
|
|
+
|
|
|
+
|
|
|
+def test_create_agent_respects_explicit_dirs(api_client, tmp_path):
|
|
|
+ client, key, ws_base, db = api_client
|
|
|
+ my_dir = str(tmp_path / "my-own-dir")
|
|
|
+ os.makedirs(my_dir)
|
|
|
+ r = client.post("/api/v1/agents", headers=_auth(key),
|
|
|
+ json={"name": "显式目录", "config": _MIN_CONFIG,
|
|
|
+ "work_dir": my_dir})
|
|
|
+ assert r.status_code in (200, 201), r.text[:300]
|
|
|
+ body = r.json()
|
|
|
+ assert body["work_dir"] == my_dir
|
|
|
+ assert body["run_dir"] == "" # 显式给了任一目录 → 其余不自动补
|
|
|
+ # 没有为它建 Workspace/<名> 目录
|
|
|
+ assert not os.path.exists(os.path.join(ws_base, "显式目录"))
|
kenny67nju