feat: agent version pinning (v4)
resolve_registry_agent in cloud/agent_def.py: agent_name may be <name> or <name>@<version>, agent_version may also be given separately; both given must agree or 400. Unpinned loads <name>.json (default); pinned loads <name>.<version>.json and requires the file's own version field to match, with no silent fallback to the default file. Version strings are restricted to alphanumerics/._- (no path separators, no ..); the agents-dir traversal guard is a second line of defence. Server: agent_version is only for agent_name (inline agent with agent_version is 400); stored agent_name is the base name, not the @ form. Snapshot fixture: agents/research-brief.v1.json (same content as the default v1 file). Tests: test_agent_version.py (7 tests).
This commit is contained in:
host committed
1 parent
8fe4ad2f38
commit
7f60f12639
4 files changed
+277
-31
No files matched your search
@@ -0,0 +1,163 @@
|
||||
"""agent 版本钉选 v4 单测：文件选择、版本一致、@ 语法、歧义拒绝。
|
||||
|
||||
运行（在仓库根目录下）：
|
||||
python3 -m unittest discover -s cloud/tests -p "test_agent_version.py" -v
|
||||
|
||||
说明：helper 部分用临时 agents 目录（不碰仓库注册表），
|
||||
服务端点部分用 create_server 本地 HTTP（test_task_options
|
||||
同款），仓库自带快照 agents/research-brief.v1.json 作
|
||||
钉选夹具。关键断言：缺快照绝不回落默认文件；文件内
|
||||
version 与请求不一致即 400；@ 与 agent_version 不一致
|
||||
即 400；inline agent 配 agent_version 即 400；钉选存库
|
||||
的 agent_name 是基本名、不是 @ 写法。
|
||||
"""
|
||||
|
||||
import http.client
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
from cloud.agent_def import list_known_skills
|
||||
from cloud.agent_def import resolve_registry_agent
|
||||
from cloud.auth import AuthStore
|
||||
from cloud.server import create_server
|
||||
except ImportError:
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||||
from cloud.agent_def import list_known_skills
|
||||
from cloud.agent_def import resolve_registry_agent
|
||||
from cloud.auth import AuthStore
|
||||
from cloud.server import create_server
|
||||
|
||||
|
||||
def _write_definition(directory, filename, name="demo", version="v1"):
|
||||
with open(os.path.join(directory, filename),
|
||||
"w", encoding="utf-8") as fh:
|
||||
json.dump({"name": name, "version": version,
|
||||
"skills": [], "tools": [],
|
||||
"defaults": {"effort": "low"}}, fh)
|
||||
|
||||
|
||||
class ResolveTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.dir = self.tmp.name
|
||||
_write_definition(self.dir, "demo.json", version="v1")
|
||||
_write_definition(self.dir, "demo.v2.json", version="v2")
|
||||
_write_definition(self.dir, "demo.v9.json", version="v1")
|
||||
|
||||
def tearDown(self):
|
||||
self.tmp.cleanup()
|
||||
|
||||
def test_unpinned_and_pinned(self):
|
||||
self.assertEqual(
|
||||
resolve_registry_agent("demo", agents_dir=self.dir).version,
|
||||
"v1")
|
||||
self.assertEqual(
|
||||
resolve_registry_agent("demo", version="v2",
|
||||
agents_dir=self.dir).version, "v2")
|
||||
self.assertEqual(
|
||||
resolve_registry_agent("demo@v2",
|
||||
agents_dir=self.dir).version, "v2")
|
||||
|
||||
def test_missing_snapshot_never_falls_back(self):
|
||||
with self.assertRaises(ValueError):
|
||||
resolve_registry_agent("demo", version="v3",
|
||||
agents_dir=self.dir)
|
||||
|
||||
def test_file_version_mismatch_rejected(self):
|
||||
with self.assertRaises(ValueError):
|
||||
resolve_registry_agent("demo", version="v9",
|
||||
agents_dir=self.dir)
|
||||
|
||||
def test_at_and_version_conflict_and_bad_version(self):
|
||||
with self.assertRaises(ValueError):
|
||||
resolve_registry_agent("demo@v2", version="v1",
|
||||
agents_dir=self.dir)
|
||||
with self.assertRaises(ValueError):
|
||||
resolve_registry_agent("demo", version="../x",
|
||||
agents_dir=self.dir)
|
||||
with self.assertRaises(ValueError):
|
||||
resolve_registry_agent("demo@", agents_dir=self.dir)
|
||||
with self.assertRaises(ValueError):
|
||||
resolve_registry_agent("demo@v2@v3",
|
||||
agents_dir=self.dir)
|
||||
|
||||
|
||||
class ServerVersionTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||||
self.auth = AuthStore(self.db_path)
|
||||
self.owner_id, self.owner_key = self.auth.create_user("owner")
|
||||
self.server = create_server(self.db_path, "127.0.0.1", 0)
|
||||
self.port = self.server.server_address[1]
|
||||
self.thread = threading.Thread(
|
||||
target=self.server.serve_forever, daemon=True)
|
||||
self.thread.start()
|
||||
status, self.project = self._req(
|
||||
"POST", "/projects", {"name": "demo"}, self.owner_key)
|
||||
self.assertEqual(status, 201)
|
||||
|
||||
def tearDown(self):
|
||||
self.server.shutdown()
|
||||
self.server.server_close()
|
||||
self.thread.join(timeout=5)
|
||||
try:
|
||||
self.auth.conn.close()
|
||||
except Exception:
|
||||
pass
|
||||
self.tmp.cleanup()
|
||||
|
||||
def _req(self, method, path, body=None, key=None):
|
||||
conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
|
||||
headers = {"Content-Type": "application/json"}
|
||||
if key:
|
||||
headers["X-API-Key"] = key
|
||||
conn.request(method, path,
|
||||
json.dumps(body) if body is not None else None, headers)
|
||||
resp = conn.getresponse()
|
||||
raw = resp.read()
|
||||
conn.close()
|
||||
try:
|
||||
data = json.loads(raw)
|
||||
except ValueError:
|
||||
data = None
|
||||
return resp.status, data
|
||||
|
||||
def _list_tasks(self):
|
||||
status, data = self._req(
|
||||
"GET", "/tasks?project_id=" + self.project["id"],
|
||||
key=self.owner_key)
|
||||
self.assertEqual(status, 200)
|
||||
return data["tasks"]
|
||||
|
||||
def test_pinned_by_field_and_at_syntax(self):
|
||||
for body in (
|
||||
{"agent_name": "research-brief", "agent_version": "v1"},
|
||||
{"agent_name": "research-brief@v1"},
|
||||
):
|
||||
status, task = self._req("POST", "/tasks", {
|
||||
"project_id": self.project["id"], "prompt": "hi",
|
||||
**body}, self.owner_key)
|
||||
self.assertEqual(status, 201, task)
|
||||
self.assertEqual(task["agent_name"], "research-brief")
|
||||
self.assertEqual(task["agent_version"], "v1")
|
||||
|
||||
def test_pinned_missing_and_inline_version_rejected(self):
|
||||
for body in (
|
||||
{"agent_name": "research-brief", "agent_version": "v999"},
|
||||
{"agent_name": "research-brief@v2",
|
||||
"agent_version": "v1"},
|
||||
{"agent": {"name": "research-brief", "version": "v1"},
|
||||
"agent_version": "v1"},
|
||||
):
|
||||
status, _data = self._req("POST", "/tasks", {
|
||||
"project_id": self.project["id"], "prompt": "hi",
|
||||
**body}, self.owner_key)
|
||||
self.assertEqual(status, 400, body)
|
||||
self.assertEqual(self._list_tasks(), [])
|
||||
Reference in new issue
Block a user