Require sandbox pass before builder deploy
All checks were successful
build / build (push) Successful in 12s

This commit is contained in:
robert
2026-06-07 22:53:57 -03:00
parent 49a41949ab
commit b285f3cdc8
4 changed files with 114 additions and 7 deletions

View File

@@ -1,6 +1,7 @@
from __future__ import annotations
import asyncio
import hashlib
import io
import json
import tarfile
@@ -12,6 +13,7 @@ from agent_builder.builder import _workspace_backend_with_grep_compat
from agent_builder.tools import (
_BUILDER_STATE_FILE,
_BUILDER_INTERNAL_PREFIX,
_WorkspaceObjectStore,
_parse_manifest_json,
_deploy_poll_url,
_deployment_drift_error,
@@ -380,6 +382,18 @@ class TemplateInitTests(unittest.TestCase):
}
for rel, content in files.items():
workspace.write_bytes(prefix + rel, content.encode("utf-8"))
source_hash = hashlib.sha256(
_tarball_workspace_dir(_WorkspaceObjectStore(workspace), prefix)
).hexdigest()
workspace.write_bytes(
prefix + _BUILDER_STATE_FILE,
json.dumps({
"last_sandbox": {
"source_hash": source_hash,
"exit_code": 0,
}
}).encode("utf-8"),
)
tools = build_tools(
ToolContext(
bucket="bucket",
@@ -429,6 +443,41 @@ class TemplateInitTests(unittest.TestCase):
self.assertEqual(dsl["entrypoint"]["module"], "agent")
self.assertEqual(dsl["entrypoint"]["class_name"], "DemoAgent")
def test_cp_deploy_tarball_refuses_without_current_sandbox_pass(self) -> None:
workspace = _FakeWorkspace()
prefix = "agents/demo-agent/"
workspace.write_bytes(prefix + "agent.py", b"print('ok')\n")
workspace.write_bytes(
prefix + "a2a.yaml",
b"name: demo-agent\nversion: 0.1.0\nentrypoint: agent:DemoAgent\n",
)
workspace.write_bytes(prefix + "requirements.txt", b"")
tools = build_tools(
ToolContext(
bucket="bucket",
settings=_settings(deploy_wait_timeout_s=0),
workspace=workspace,
cp_jwt="jwt-user",
)
)
deploy = _tool_by_name(tools, "cp_deploy_tarball")
_FakeAsyncClient.posts = []
result = json.loads(
asyncio.run(
deploy.ainvoke(
{
"name": "demo-agent",
"version": "0.1.0",
"public": False,
}
)
)
)
self.assertEqual(result["error"], "sandbox_not_passed")
self.assertEqual(_FakeAsyncClient.posts, [])
def test_cp_compose_meta_agent_posts_manifest_and_records_state(self) -> None:
workspace = _FakeWorkspace()
tools = build_tools(