fix agent template init workspace writes
All checks were successful
build / build (push) Successful in 2s
All checks were successful
build / build (push) Successful in 2s
This commit is contained in:
@@ -1,9 +1,11 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import json
|
||||
import tarfile
|
||||
import unittest
|
||||
|
||||
from agent_builder.config import Settings
|
||||
from agent_builder.tools import (
|
||||
_BUILDER_STATE_FILE,
|
||||
_BUILDER_INTERNAL_PREFIX,
|
||||
@@ -14,6 +16,8 @@ from agent_builder.tools import (
|
||||
_render_a2a_init_template,
|
||||
_render_skill_md,
|
||||
_tarball_workspace_dir,
|
||||
ToolContext,
|
||||
build_tools,
|
||||
)
|
||||
|
||||
|
||||
@@ -205,6 +209,30 @@ class TemplateInitTests(unittest.TestCase):
|
||||
],
|
||||
)
|
||||
|
||||
def test_init_agent_template_writes_through_workspace_store(self) -> None:
|
||||
workspace = _FakeWorkspace()
|
||||
tools = build_tools(
|
||||
ToolContext(
|
||||
bucket="bucket",
|
||||
settings=_settings(),
|
||||
workspace=workspace,
|
||||
)
|
||||
)
|
||||
|
||||
init = _tool_by_name(tools, "init_agent_template")
|
||||
result = json.loads(
|
||||
init.invoke({
|
||||
"name": "website-scraper",
|
||||
"description": "Browser scraper",
|
||||
"frontend": "none",
|
||||
})
|
||||
)
|
||||
|
||||
self.assertTrue(result["ok"])
|
||||
self.assertIn("agents/website-scraper/agent.py", workspace.objects)
|
||||
self.assertIn("agents/website-scraper/a2a.yaml", workspace.objects)
|
||||
self.assertIn("agents/website-scraper/requirements.txt", workspace.objects)
|
||||
|
||||
|
||||
def _tarball(files: dict[str, str]) -> bytes:
|
||||
buf = io.BytesIO()
|
||||
@@ -259,3 +287,44 @@ class _FakePaginator:
|
||||
]
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
class _FakeWorkspace:
|
||||
def __init__(self) -> None:
|
||||
self.objects: dict[str, bytes] = {}
|
||||
|
||||
def iter_paths(self) -> list[str]:
|
||||
return sorted(self.objects)
|
||||
|
||||
def read_bytes(self, key: str) -> bytes:
|
||||
return self.objects[key]
|
||||
|
||||
def write_bytes(self, key: str, body: bytes) -> None:
|
||||
self.objects[key] = bytes(body)
|
||||
|
||||
def delete_path(self, key: str) -> None:
|
||||
self.objects.pop(key, None)
|
||||
|
||||
|
||||
def _tool_by_name(tools: list[object], name: str) -> object:
|
||||
for tool in tools:
|
||||
if getattr(tool, "name", None) == name:
|
||||
return tool
|
||||
raise AssertionError(f"tool not found: {name}")
|
||||
|
||||
|
||||
def _settings() -> Settings:
|
||||
return Settings(
|
||||
sandbox_url="http://sandbox.test",
|
||||
sandbox_timeout_s=1,
|
||||
sandbox_token=None,
|
||||
deploy_wait_timeout_s=1,
|
||||
litellm_url="http://litellm.test",
|
||||
litellm_key="",
|
||||
litellm_model="gpt-test",
|
||||
cp_url="http://cp.test",
|
||||
minio_endpoint="http://minio.test",
|
||||
minio_access_key="key",
|
||||
minio_secret_key="secret",
|
||||
image="python:3.11-slim",
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user