feat(workspace): port cwd across backends

This commit is contained in:
2026-06-02 17:01:19 +00:00
parent 6150497b47
commit d5fcbe53ef
10 changed files with 229 additions and 31 deletions
+50
View File
@@ -30,6 +30,7 @@ from bot_bottle.backend.smolmachines.provision import (
provider_auth as _provider_auth,
skills as _skills,
supervise as _supervise,
workspace as _workspace,
)
from bot_bottle.backend.smolmachines.launch import _bundle_launch_spec
from bot_bottle.backend.smolmachines.smolvm import SmolvmRunResult
@@ -848,6 +849,55 @@ class TestProvisionGitUser(unittest.TestCase):
self.assertEqual(["user.email", "bot@example.com"], calls[0][0][7:])
class TestProvisionWorkspace(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory(prefix="cb-prov-workspace.")
self.stage = Path(self._tmp.name)
def tearDown(self):
self._tmp.cleanup()
def test_noop_when_copy_cwd_false(self):
plan = _plan(copy_cwd=False, stage_dir=self.stage)
with patch(
"bot_bottle.backend.smolmachines.provision.workspace._smolvm.machine_cp"
) as cp, patch(
"bot_bottle.backend.smolmachines.provision.workspace._smolvm.machine_exec"
) as ex:
_workspace.provision_workspace(plan, "bot-bottle-demo-abc12")
cp.assert_not_called()
ex.assert_not_called()
def test_copies_workspace_to_plan_path_and_chowns(self):
cwd = self.stage / "cwd"
cwd.mkdir()
plan = _plan(copy_cwd=True, user_cwd=str(cwd), stage_dir=self.stage)
with patch(
"bot_bottle.backend.smolmachines.provision.workspace._smolvm.machine_cp"
) as cp, patch(
"bot_bottle.backend.smolmachines.provision.workspace._smolvm.machine_exec"
) as ex:
_workspace.provision_workspace(plan, "bot-bottle-demo-abc12")
cp.assert_called_once_with(
str(cwd),
"bot-bottle-demo-abc12:/home/node/workspace",
)
argvs = [c.args[1] for c in ex.call_args_list]
self.assertIn(
["sh", "-c", "rm -rf /home/node/workspace && mkdir -p /home/node"],
argvs,
)
self.assertIn(
[
"sh", "-c",
"chown -R node:node /home/node/workspace && "
"chmod 755 /home/node/workspace",
],
argvs,
)
class TestProvisionSupervise(unittest.TestCase):
def test_noop_when_supervise_not_enabled(self):
with patch(