From b16bb31d244d28def8e1595f923463ae5bc5035a Mon Sep 17 00:00:00 2001 From: Juan Diaz Date: Thu, 23 Jul 2026 02:57:56 +0200 Subject: [PATCH] Use SDK for preset VM creation --- docs/build-local-debian-pi-image.md | 8 +- docs/fragile-glue.md | 15 + docs/per-mount-readonly-writable-draft.md | 8 +- docs/pi-package-node-version-issue.md | 8 +- src/sbx/cli.py | 340 +++---------- src/sbx/completion.py | 2 - src/sbx/image/build_debian.py | 133 +----- src/sbx/smolvm_preset.py | 80 ++++ src/sbx/vm_state.py | 14 - tests/test_build_debian_image.py | 142 +----- tests/test_cli.py | 550 +++++++--------------- tests/test_cli_extra.py | 15 +- tests/test_smolvm_preset.py | 208 ++++++++ 13 files changed, 562 insertions(+), 961 deletions(-) create mode 100644 src/sbx/smolvm_preset.py create mode 100644 tests/test_smolvm_preset.py diff --git a/docs/build-local-debian-pi-image.md b/docs/build-local-debian-pi-image.md index 2d2bdbb..ca8458e 100644 --- a/docs/build-local-debian-pi-image.md +++ b/docs/build-local-debian-pi-image.md @@ -84,13 +84,7 @@ Or pass a fully composed Containerfile directly: sbx image build-debian --containerfile path/to/Containerfile ``` -By default the subcommand prints only the built image paths and a minimal `sbx` config snippet. To also print a SmolVM SDK usage sketch after building, pass `--sdk-sketch`. - -To print the SDK sketch later without rebuilding the image, run: - -```bash -sbx image build-debian --print-sdk-sketch ~/.smolvm/images/debian-sbx -``` +The subcommand prints the built image paths and a minimal `sbx` config snippet. The subcommand also writes a local image manifest: diff --git a/docs/fragile-glue.md b/docs/fragile-glue.md index 356a1ef..07f4e53 100644 --- a/docs/fragile-glue.md +++ b/docs/fragile-glue.md @@ -17,3 +17,18 @@ Why: SmolVM has an internal `custom_commands` hook in `_base_init_script()`, but Fragility: `_default_init_script()` and `_base_init_script()` are protected SmolVM internals and may change. Exit: replace with a public SmolVM boot-hook API, e.g. `/etc/smolvm/boot.d/*.sh` or a public `custom_commands`/`boot_hooks` parameter. + +## SmolVM preset creation through private facade methods + +Used by `src/sbx/smolvm_preset.py`: + +```python +from smolvm.facade import _build_auto_config +channel = vm._ensure_ssh_for_env() +``` + +Why: SmolVM 0.0.28 has public VM lifecycle and preset application APIs, but no public operation that creates a named auto-configured VM with sbx's CPU/port-forward settings and supplies the communication channel required by `apply_preset()`. The preset API also reads `os.environ`, so sbx temporarily activates its credential-filtered environment during provisioning. + +Fragility: both methods are private and may change. The SmolVM dependency remains pinned to `smolvm==0.0.28`, and all preset-specific private access is contained in `smolvm_preset.py`. + +Exit: replace this module with an upstream public preset creation API that accepts explicit VM resources, mounts, port forwards, timeouts, and host credential/environment inputs; then remove this entry and the adjacent `ponytail:` comment. diff --git a/docs/per-mount-readonly-writable-draft.md b/docs/per-mount-readonly-writable-draft.md index 8487eda..6974b04 100644 --- a/docs/per-mount-readonly-writable-draft.md +++ b/docs/per-mount-readonly-writable-draft.md @@ -122,13 +122,9 @@ Estimated difficulty: medium. `sbx` could construct `WorkspaceMount` objects directly and use SmolVM's Python API for all normal starts. -This would allow per-mount control without changing SmolVM CLI, but it is more invasive for `sbx` because the normal code path currently shells out to preset commands such as: +Preset-backed VM creation now goes through sbx's SmolVM SDK compatibility module, so this option no longer requires replacing a preset CLI subprocess. It still requires sbx to model per-mount writability and pass `WorkspaceMount` objects into that creation boundary. -```bash -smolvm pi start ... -``` - -Estimated difficulty: medium/high compared to extending SmolVM CLI. +Estimated difficulty: medium compared to extending SmolVM CLI. ## Suggested future `sbx` interface diff --git a/docs/pi-package-node-version-issue.md b/docs/pi-package-node-version-issue.md index f8b89f0..987654d 100644 --- a/docs/pi-package-node-version-issue.md +++ b/docs/pi-package-node-version-issue.md @@ -76,13 +76,7 @@ That package points to the older GitHub organization/account: badlogic/pi-mono ``` -`sbx` currently delegates agent installation to SmolVM: - -```bash -smolvm pi start -``` - -So this stale package likely comes from SmolVM's Pi preset or prebuilt Pi image, not directly from `sbx`. +`sbx` delegates agent installation to SmolVM's Pi preset through its SDK compatibility layer. Therefore this stale package likely comes from the SmolVM preset or prebuilt Pi image, not directly from `sbx`. ### 2. New Pi requires newer Node than the VM provides diff --git a/src/sbx/cli.py b/src/sbx/cli.py index 4cc3eac..225c614 100644 --- a/src/sbx/cli.py +++ b/src/sbx/cli.py @@ -6,8 +6,8 @@ import sys import tempfile import tomllib -from collections.abc import Iterator, Mapping, Sequence -from contextlib import contextmanager, suppress +from collections.abc import Mapping, Sequence +from contextlib import suppress from pathlib import Path from typing import Any @@ -20,6 +20,7 @@ lifecycle_warnings, runtime, session_state, + smolvm_preset, vm_metadata, vm_state, ) @@ -253,18 +254,6 @@ def _sync_forwarded_env_or_error(vm_id: str, names: list[str]) -> bool: return True -@contextmanager -def _patched_environ(env: Mapping[str, str]) -> Iterator[None]: - original = dict(os.environ) - os.environ.clear() - os.environ.update(env) - try: - yield - finally: - os.environ.clear() - os.environ.update(original) - - def _validate_cpus(value: Any) -> int: cpus = int(value) if not 1 <= cpus <= 32: @@ -380,38 +369,6 @@ def _start_existing_vm_if_needed(vm_id: str, status: str, boot_timeout: float) - return rc -def _smolvm_error_message(stdout: str, fallback: str) -> str: - try: - payload = json.loads(stdout) - message = payload["error"]["message"] - except (KeyError, TypeError, json.JSONDecodeError): - return fallback - return message if isinstance(message, str) and message else fallback - - -def _print_start_failure(stdout: str) -> None: - message = _smolvm_error_message(stdout, "SmolVM failed to start the VM.") - print(f"sbx: {message}", file=sys.stderr) - if "QEMU exited early" in message: - print( - "sbx: The VM was created but the backend failed during boot. " - "Try `sbx recreate --force`; if it repeats, run `sbx doctor` " - "and inspect the SmolVM backend runtime logs.", - file=sys.stderr, - ) - - -def _extract_started_vm_name(stdout: str) -> str: - try: - payload = json.loads(stdout) - name = payload["data"]["vm"]["name"] - except (KeyError, TypeError, json.JSONDecodeError) as exc: - raise ConfigError("could not read VM name from smolvm JSON output") from exc - if not isinstance(name, str) or not name: - raise ConfigError("smolvm JSON output did not include a VM name") - return name - - def _stop_vm_if_last_session(vm_id: str, *, stop_on_exit: bool) -> None: if not stop_on_exit: runtime.debug(f"not stopping {vm_id}: stop_on_exit disabled") @@ -590,119 +547,6 @@ def _maybe_write_project_config( ) -def _start_preset_with_sdk( - *, - args: argparse.Namespace, - config: Mapping[str, Any], - agent: str, - cpus: int | None, - mounts: Sequence[str], - writable_mounts: bool, - attach: bool, - run_user: str | None, - auth_port: bool, - auth_host_port: int, - auth_guest_port: int, - stop_on_exit: bool, - cwd: str | None, - git_config_text: str | None, - copy_host_credentials: bool, - forward_env: list[str], - json_output: bool, - port_forwards: Sequence[str], -) -> int: - from smolvm import SmolVM - from smolvm.facade import _build_auto_config - from smolvm.presets import apply_preset, get_preset - from smolvm.types import PortForwardConfig - - preset_name = "claude-code" if agent == "claude" else agent - preset = get_preset(preset_name) - memory = _arg_or_config(args, "memory", config, "sbx", default=preset.default_mem_mib) - disk_size = _arg_or_config(args, "disk_size", config, "sbx", default=preset.default_disk_mib) - requested_name = _arg_or_config(args, "name", config, "sbx") - requested_os = _arg_or_config(args, "os", config, "sbx", default="ubuntu") - install_timeout = int(_arg_or_config(args, "install_timeout", config, "sbx", default=600)) - boot_timeout = _validate_boot_timeout( - _arg_or_config(args, "boot_timeout", config, "sbx", default=DEFAULT_BOOT_TIMEOUT) - ) - - def start() -> str: - vm = None - config_obj, ssh_key_path = _build_auto_config( - vm_name=str(requested_name) if requested_name else None, - name_prefix=agent, - os=str(requested_os), - backend=DEFAULT_BACKEND, - memory=int(memory), - disk_size_mib=int(disk_size), - ssh_key_path=None, - ) - updates: dict[str, Any] = { - "port_forwards": [ - PortForwardConfig(**item) - for item in network.port_forwards_from_specs(port_forwards) - ] - } - if cpus is not None: - updates["vcpu_count"] = cpus - config_obj = config_obj.model_copy(update=updates) - try: - vm = SmolVM( - config_obj, - ssh_key_path=ssh_key_path, - mounts=list(mounts), - writable_mounts=writable_mounts, - ) - vm.start(boot_timeout=boot_timeout) - vm.wait_for_ssh(timeout=boot_timeout) - guest_setup.set_hostname(str(vm.vm_id), run_capture=runtime.run_capture) - ssh = vm._ensure_ssh_for_env() - apply_preset(ssh, preset, install_timeout=install_timeout) - return str(vm.vm_id) - finally: - if vm is not None: - vm.close() - - env = guest_setup.sanitize_forwarded_env(dict(os.environ), forward_env) - temp_home_ctx = None - if not copy_host_credentials: - temp_home_ctx = tempfile.TemporaryDirectory(prefix="sbx-no-credentials-") - env = guest_setup.credential_free_env(Path(temp_home_ctx.name), forward_env=forward_env) - runtime.debug(f"credential-free HOME: {temp_home_ctx.name}") - - try: - with _patched_environ(env): - vm_name = start() - finally: - if temp_home_ctx is not None: - temp_home_ctx.cleanup() - - _maybe_write_project_config(args, config, vm_name=vm_name, agent=agent, created=True) - vm_metadata.record_vm_project(vm_name, _project_identity(args, config)) - - if json_output: - print(json.dumps({"vm": {"name": vm_name, "status": "running"}})) - elif attach: - user_msg = f" as user {run_user}" if run_user is not None else "" - print(f"Started '{vm_name}'. Launching {agent}{user_msg}...") - else: - print(f"Started '{vm_name}'.") - - return _post_start_actions( - vm_name=vm_name, - command=LAUNCH_COMMANDS[agent], - attach=attach, - run_user=run_user, - auth_port=auth_port, - auth_host_port=auth_host_port, - auth_guest_port=auth_guest_port, - stop_on_exit=stop_on_exit, - cwd=cwd, - git_config_text=git_config_text, - ) - - def _start_local_image( *, args: argparse.Namespace, @@ -864,28 +708,24 @@ def cmd_start(args: argparse.Namespace) -> int: print(f"sbx: agent must be one of: {', '.join(AGENTS)}", file=sys.stderr) return 2 - argv: list[str] = [str(agent), "start"] configured_backend = _cfg(config, "sbx", "backend", DEFAULT_BACKEND) if configured_backend != DEFAULT_BACKEND: print("sbx: other backends are not supported yet", file=sys.stderr) return 2 - scalar_options = ( - ("name", "--name"), - ("memory", "--memory"), - ("disk_size", "--disk-size"), - ("os", "--os"), - ("install_timeout", "--install-timeout"), - ) - for attr, flag in scalar_options: - value = _arg_or_config(args, attr, config, "sbx") - if value is not None: - argv += [flag, str(value)] - argv += ["--backend", DEFAULT_BACKEND] + memory_value = _arg_or_config(args, "memory", config, "sbx") + disk_size_value = _arg_or_config(args, "disk_size", config, "sbx") + install_timeout_value = _arg_or_config(args, "install_timeout", config, "sbx", default=600) + try: + memory_mib = int(memory_value) if memory_value is not None else None + disk_size_mib = int(disk_size_value) if disk_size_value is not None else None + install_timeout = int(install_timeout_value) + except (TypeError, ValueError) as exc: + raise ConfigError("memory, disk_size, and install_timeout must be integers") from exc + requested_os = str(_arg_or_config(args, "os", config, "sbx", default="ubuntu")) boot_timeout = _validate_boot_timeout( _arg_or_config(args, "boot_timeout", config, "sbx", default=DEFAULT_BOOT_TIMEOUT) ) - argv += ["--boot-timeout", f"{boot_timeout:g}"] mounts = ( args.mount @@ -897,20 +737,14 @@ def cmd_start(args: argparse.Namespace) -> int: project_path = _arg_or_config(args, "project_path", config, "sbx") project_guest_cwd = _project_guest_cwd(project_path) if project_path is not None: - project_mount = _same_path_mount(str(project_path)) - effective_mounts.append(project_mount) - argv += ["--mount", project_mount] + effective_mounts.append(_same_path_mount(str(project_path))) for mount in mounts or []: - mount = mount if ":" in mount else _same_path_mount(mount) - effective_mounts.append(mount) - argv += ["--mount", mount] + effective_mounts.append(mount if ":" in mount else _same_path_mount(mount)) writable_mounts = bool(_arg_or_config(args, "writable_mounts", config, "sbx", default=False)) if project_path is not None: writable_mounts = True - if writable_mounts: - argv.append("--writable-mounts") attach = True if args.attach is None else bool(args.attach) run_user = _arg_or_config(args, "run_user", config, "sbx") @@ -939,7 +773,7 @@ def cmd_start(args: argparse.Namespace) -> int: cpus = _validate_cpus(cpus_value) if cpus_value is not None else None try: port_forwards = _list_value(sbx_cfg.get("port_forwards"), key="[sbx].port_forwards") - network.port_forwards_from_specs(port_forwards) + parsed_port_forwards = network.port_forwards_from_specs(port_forwards) except ConfigError as exc: print(f"sbx: {exc}", file=sys.stderr) return 2 @@ -1026,112 +860,64 @@ def cmd_start(args: argparse.Namespace) -> int: print(f"sbx: failed to start image: {exc}", file=sys.stderr) return 1 - if cpus is not None or port_forwards: - try: - return _start_preset_with_sdk( - args=args, - config=config, - agent=str(agent), - cpus=cpus, - mounts=effective_mounts, - writable_mounts=writable_mounts, - attach=attach, - run_user=str(run_user) if run_user is not None else None, - auth_port=auth_port, - auth_host_port=auth_host_port, - auth_guest_port=auth_guest_port, - stop_on_exit=stop_on_exit, - cwd=project_guest_cwd, - git_config_text=git_config_text, - copy_host_credentials=copy_host_credentials, - forward_env=forward_env, - json_output=bool(args.json), - port_forwards=port_forwards, - ) - except ConfigError as exc: - print(f"sbx: {exc}", file=sys.stderr) - return 2 - except Exception as exc: # noqa: BLE001 - keep CLI errors user-friendly. - if _maybe_print_boot_timeout_running_hint( - str(requested_name) if requested_name else None, boot_timeout - ): - return 1 - print(f"sbx: failed to start preset with cpus={cpus}: {exc}", file=sys.stderr) - return 1 - - managed_start = auth_port or ( - attach - and (run_user is not None or project_guest_cwd is not None or git_config_text is not None) - ) - argv.append("--no-attach" if managed_start or not attach else "--attach") - - json_output = bool(args.json) - if json_output or managed_start: - argv.append("--json") - temp_home_ctx = None - smolvm_env = guest_setup.sanitize_forwarded_env(dict(os.environ), forward_env) + host_env = guest_setup.sanitize_forwarded_env(dict(os.environ), forward_env) if not copy_host_credentials: temp_home_ctx = tempfile.TemporaryDirectory(prefix="sbx-no-credentials-") - smolvm_env = guest_setup.credential_free_env( + host_env = guest_setup.credential_free_env( Path(temp_home_ctx.name), forward_env=forward_env ) runtime.debug(f"credential-free HOME: {temp_home_ctx.name}") - if not managed_start: - try: - rc = runtime.run_smolvm(argv, env=smolvm_env) - if rc == 0 and requested_name: - guest_setup.set_hostname(str(requested_name), run_capture=runtime.run_capture) - _maybe_write_project_config( - args, config, vm_name=str(requested_name), agent=str(agent), created=True - ) - vm_metadata.record_vm_project(str(requested_name), project_identity) - return rc - finally: - if temp_home_ctx is not None: - temp_home_ctx.cleanup() - - completed = runtime.run_smolvm_capture(argv, env=smolvm_env) - if temp_home_ctx is not None: - temp_home_ctx.cleanup() - if completed is None: - return 127 - if completed.stderr: - print(completed.stderr, end="", file=sys.stderr) - if completed.returncode != 0: - if json_output and completed.stdout: - print(completed.stdout, end="") - elif completed.stdout: - _print_start_failure(completed.stdout) - _maybe_print_boot_timeout_running_hint( - str(requested_name) if requested_name else None, boot_timeout + vm = None + try: + vm = smolvm_preset.create_preset( + str(agent), + vm_name=str(requested_name) if requested_name else None, + guest_os=requested_os, + cpus=cpus, + memory_mib=memory_mib, + disk_size_mib=disk_size_mib, + mounts=effective_mounts, + writable_mounts=writable_mounts, + port_forwards=parsed_port_forwards, + boot_timeout=boot_timeout, + install_timeout=install_timeout, + host_env=host_env, ) - return completed.returncode + vm_name = str(vm.vm_id) + guest_setup.set_hostname(vm_name, run_capture=runtime.run_capture) + except Exception as exc: # noqa: BLE001 - keep CLI errors user-friendly. + if _maybe_print_boot_timeout_running_hint( + str(requested_name) if requested_name else None, boot_timeout + ): + return 1 + print(f"sbx: failed to create preset {str(agent)!r}: {exc}", file=sys.stderr) + return 1 + finally: + if vm is not None: + with suppress(Exception): + vm.close() + if temp_home_ctx is not None: + temp_home_ctx.cleanup() - vm_name = _extract_started_vm_name(completed.stdout) - guest_setup.set_hostname(vm_name, run_capture=runtime.run_capture) _maybe_write_project_config(args, config, vm_name=vm_name, agent=str(agent), created=True) vm_metadata.record_vm_project(vm_name, project_identity) - if auth_port: - port_rc = network.expose_auth_port(vm_name, auth_host_port, auth_guest_port) - if port_rc != 0: - return port_rc - if json_output: - print(completed.stdout, end="") + if args.json: + print(json.dumps({"vm": {"name": vm_name, "status": "running"}})) elif attach: user_msg = f" as user {run_user}" if run_user is not None else "" print(f"Started '{vm_name}'. Launching {agent}{user_msg}...") else: - print(f"Started '{vm_name}'. Auth callback port: localhost:{auth_host_port}") + print(f"Started '{vm_name}'.") return _post_start_actions( vm_name=vm_name, command=LAUNCH_COMMANDS[str(agent)], attach=attach, run_user=str(run_user) if run_user is not None else None, - auth_port=False, + auth_port=auth_port, auth_host_port=auth_host_port, auth_guest_port=auth_guest_port, stop_on_exit=stop_on_exit, @@ -1615,33 +1401,13 @@ def build_parser() -> argparse.ArgumentParser: return parser -def _normalize_argv(argv: Sequence[str]) -> list[str]: - """Normalize `sbx run NAME --flag` to `sbx run --name NAME --flag`.""" - normalized = list(argv) - for command in ("run", "create", "recreate"): - if command not in normalized: - continue - command_index = normalized.index(command) - name_index = command_index + 1 - if name_index >= len(normalized): - return normalized - candidate = normalized[name_index] - if candidate != "--" and not candidate.startswith("-"): - normalized.insert(name_index, "--name") - return normalized - return normalized - - def main(argv: Sequence[str] | None = None) -> int: os.environ.setdefault("PYTHONUNBUFFERED", "1") raw_argv = list(argv) if argv is not None else sys.argv[1:] - normalized_argv = _normalize_argv(raw_argv) parser = build_parser() - args = parser.parse_args(normalized_argv) + args = parser.parse_args(raw_argv) runtime.DEBUG = bool(args.debug) runtime.debug(f"argv: {raw_argv}") - if normalized_argv != raw_argv: - runtime.debug(f"normalized argv: {normalized_argv}") if args.action in {"completion", "image"}: args.config_data = {} else: diff --git a/src/sbx/completion.py b/src/sbx/completion.py index 547d05a..9ed125d 100644 --- a/src/sbx/completion.py +++ b/src/sbx/completion.py @@ -81,8 +81,6 @@ "--cache-dir", "--kernel-url", "--json", - "--sdk-sketch", - "--print-sdk-sketch", "--help", ) IMAGE_LS_OPTIONS = ("--json", "--help") diff --git a/src/sbx/image/build_debian.py b/src/sbx/image/build_debian.py index 2f2fffc..8740de4 100644 --- a/src/sbx/image/build_debian.py +++ b/src/sbx/image/build_debian.py @@ -17,10 +17,12 @@ import urllib.request from collections.abc import Iterator from contextlib import contextmanager -from importlib.resources import files -from importlib.resources.abc import Traversable +from importlib.resources import as_file, files from pathlib import Path +from smolvm.images.builder import ImageBuilder +from smolvm.images.published import BASE_KERNELS + RESOURCE_PACKAGE = "sbx.image.resources" DEFAULT_BASE_CONTAINERFILE = Path("Containers") / "Debian" / "Base.Containerfile" DEFAULT_AGENT_CONTAINERFILE = Path("Containers") / "Agents" / "Pi.Containerfile" @@ -40,23 +42,9 @@ DOCKER_KERNEL_NAME = "vmlinux-docker.bin" -def _copy_resource_tree(source: Traversable, target: Path) -> None: - if source.is_dir(): - target.mkdir(parents=True, exist_ok=True) - for child in source.iterdir(): - _copy_resource_tree(child, target / child.name) - else: - target.parent.mkdir(parents=True, exist_ok=True) - target.write_bytes(source.read_bytes()) - - @contextmanager def _packaged_resources() -> Iterator[Path]: - with tempfile.TemporaryDirectory(prefix="sbx-image-build-resources-") as tmp: - root = Path(tmp) - package_root = files(RESOURCE_PACKAGE) - for name in ("Containers", "kernel", "scripts"): - _copy_resource_tree(package_root / name, root / name) + with as_file(files(RESOURCE_PACKAGE)) as root: yield root @@ -94,20 +82,17 @@ def _compose_containerfiles( output.write_text(content, encoding="utf-8") -def _docker_builder_class(builder_class: type) -> type: - class SbxDockerImageBuilder(builder_class): # type: ignore[misc, valid-type] - def _default_init_script(self) -> str: - # ponytail: protected SmolVM hook; replace with public boot hooks when upstream exists. - return self._base_init_script( - custom_commands=""" +class SbxDockerImageBuilder(ImageBuilder): + def _default_init_script(self) -> str: + # ponytail: protected SmolVM hook; replace with public boot hooks when upstream exists. + return self._base_init_script( + custom_commands=""" # sbx: start rootless Docker at boot for Docker-capable images. if [ -x /usr/local/bin/sbx-start-rootless-docker ]; then /usr/local/bin/sbx-start-rootless-docker >/var/log/sbx-rootless-docker.log 2>&1 & fi """ - ) - - return SbxDockerImageBuilder + ) def _download(url: str, output: Path) -> None: @@ -254,62 +239,6 @@ def _build_containerfile_base_image( return root_tag -def _print_sdk_sketch(*, kernel_path: Path, rootfs_path: Path, boot_args: str) -> None: - print("SDK usage sketch:") - print(" from pathlib import Path") - print(" from smolvm import SmolVM, VMConfig") - print(" config = VMConfig(") - print(" vm_id='debian-test',") - print(f" kernel_path=Path({str(kernel_path)!r}),") - print(f" rootfs_path=Path({str(rootfs_path)!r}),") - print(f" boot_args={boot_args!r},") - print(" backend='qemu',") - print(" ssh_capable=True,") - print(" )") - print(" with SmolVM(config) as vm:") - print(" vm.start()") - print(" vm.wait_for_ssh()") - print(" print(vm.run('cat /etc/debian_version'))") - - -def _print_existing_sdk_sketch(image_dir: Path) -> int: - image_dir = image_dir.expanduser().resolve() - manifest_path = image_dir / "smolvm-image.json" - try: - manifest = json.loads(manifest_path.read_text(encoding="utf-8")) - kernel_name = manifest["kernel"] - rootfs_name = manifest["rootfs"] - except FileNotFoundError: - print(f"build-debian-image: manifest not found: {manifest_path}", file=sys.stderr) - return 2 - except (KeyError, TypeError, json.JSONDecodeError) as exc: - print(f"build-debian-image: invalid image manifest {manifest_path}: {exc}", file=sys.stderr) - return 2 - if not isinstance(kernel_name, str) or not isinstance(rootfs_name, str): - print( - f"build-debian-image: invalid image manifest {manifest_path}: " - "kernel and rootfs must be strings", - file=sys.stderr, - ) - return 2 - boot_args = manifest.get( - "boot_args", "console=ttyS0 reboot=k panic=1 pci=off root=/dev/vda rw init=/init" - ) - if not isinstance(boot_args, str): - print( - f"build-debian-image: invalid image manifest {manifest_path}: " - "boot_args must be a string", - file=sys.stderr, - ) - return 2 - _print_sdk_sketch( - kernel_path=image_dir / kernel_name, - rootfs_path=image_dir / rootfs_name, - boot_args=boot_args, - ) - return 0 - - def add_arguments(parser: argparse.ArgumentParser) -> None: parser.add_argument( "--name", @@ -378,29 +307,9 @@ def add_arguments(parser: argparse.ArgumentParser) -> None: action="store_true", help="Print machine-readable JSON instead of a text summary.", ) - parser.add_argument( - "--sdk-sketch", - action="store_true", - help="Include a SmolVM SDK usage sketch in the text summary.", - ) - parser.add_argument( - "--print-sdk-sketch", - type=Path, - metavar="IMAGE_DIR", - help="Print a SmolVM SDK usage sketch for an existing local image directory and exit.", - ) - - -def _build_parser() -> argparse.ArgumentParser: - parser = argparse.ArgumentParser(description="Build a local Debian SSH-ready image for SmolVM.") - add_arguments(parser) - return parser def main_from_args(args: argparse.Namespace) -> int: - if args.print_sdk_sketch is not None: - return _print_existing_sdk_sketch(args.print_sdk_sketch) - if args.with_docker and args.containerfile is not None: print( "build-debian-image: --with-docker cannot be combined with --containerfile; " @@ -416,19 +325,7 @@ def main_from_args(args: argparse.Namespace) -> int: ) return 2 - try: - from smolvm.images.builder import ImageBuilder - from smolvm.images.published import BASE_KERNELS - except ImportError as exc: - print( - "build-debian-image: smolvm is not installed. Install the supported tool, " - "for example: uv tool install 'smolvm==0.0.19'", - file=sys.stderr, - ) - print(f"Original import error: {exc}", file=sys.stderr) - return 127 - - BuilderClass = _docker_builder_class(ImageBuilder) if args.with_docker else ImageBuilder + BuilderClass = SbxDockerImageBuilder if args.with_docker else ImageBuilder builder = BuilderClass(cache_dir=args.cache_dir.expanduser() if args.cache_dir else None) base_image = args.base_image arch = "amd64" if builder._host_arch_key() == "x86_64" else "arm64" @@ -535,14 +432,12 @@ def main_from_args(args: argparse.Namespace) -> int: print(" [sbx]") print(f" image = {str(manifest_path.parent)!r}") print(" run_user = 'agent'") - if args.sdk_sketch: - print() - _print_sdk_sketch(kernel_path=kernel_path, rootfs_path=rootfs_path, boot_args=boot_args) return 0 def main(argv: list[str] | None = None) -> int: - parser = _build_parser() + parser = argparse.ArgumentParser(description="Build a local Debian SSH-ready image for SmolVM.") + add_arguments(parser) return main_from_args(parser.parse_args(argv)) diff --git a/src/sbx/smolvm_preset.py b/src/sbx/smolvm_preset.py new file mode 100644 index 0000000..708a97d --- /dev/null +++ b/src/sbx/smolvm_preset.py @@ -0,0 +1,80 @@ +"""Temporary in-process SmolVM preset creation compatibility layer.""" + +import os +from collections.abc import Iterator, Mapping, Sequence +from contextlib import contextmanager, suppress + +from smolvm import SmolVM +from smolvm.facade import _build_auto_config +from smolvm.presets import apply_preset, get_preset +from smolvm.types import PortForwardConfig + +from sbx.constants import DEFAULT_BACKEND + +# ponytail: private SmolVM seams; remove when upstream exposes preset creation. + + +@contextmanager +def _environment(values: Mapping[str, str]) -> Iterator[None]: + original = dict(os.environ) + os.environ.clear() + os.environ.update(values) + try: + yield + finally: + os.environ.clear() + os.environ.update(original) + + +def create_preset( + preset_name: str, + *, + vm_name: str | None, + guest_os: str, + cpus: int | None, + memory_mib: int | None, + disk_size_mib: int | None, + mounts: Sequence[str], + writable_mounts: bool, + port_forwards: Sequence[Mapping[str, object]], + boot_timeout: float, + install_timeout: int, + host_env: Mapping[str, str], +) -> SmolVM: + """Create and provision a running preset VM; the caller must close the facade.""" + preset = get_preset("claude-code" if preset_name == "claude" else preset_name) + memory_mib = preset.default_mem_mib if memory_mib is None else memory_mib + disk_size_mib = preset.default_disk_mib if disk_size_mib is None else disk_size_mib + + with _environment(host_env): + config, ssh_key_path = _build_auto_config( + vm_name=vm_name, + name_prefix=preset_name, + os=guest_os, + backend=DEFAULT_BACKEND, + memory=memory_mib, + disk_size_mib=disk_size_mib, + ssh_key_path=None, + ) + updates: dict[str, object] = { + "port_forwards": [PortForwardConfig(**value) for value in port_forwards] + } + if cpus is not None: + updates["vcpu_count"] = cpus + config = config.model_copy(update=updates) + vm = SmolVM( + config, + ssh_key_path=ssh_key_path, + mounts=list(mounts), + writable_mounts=writable_mounts, + ) + try: + vm.start(boot_timeout=boot_timeout) + vm.wait_for_ssh(timeout=boot_timeout) + channel = vm._ensure_ssh_for_env() + apply_preset(channel, preset, install_timeout=install_timeout) + except BaseException: + with suppress(Exception): + vm.close() + raise + return vm diff --git a/src/sbx/vm_state.py b/src/sbx/vm_state.py index f18c85a..10a05e7 100644 --- a/src/sbx/vm_state.py +++ b/src/sbx/vm_state.py @@ -1,6 +1,5 @@ import json import sqlite3 -from pathlib import Path from typing import Any from sbx.constants import SMOLVM_DB_PATH @@ -39,16 +38,3 @@ def mark_error_vm_stopped_for_restart(vm_id: str) -> None: """, (vm_id,), ) - - -def vm_image(vm: Any) -> str: - rootfs = getattr(getattr(vm, "config", None), "rootfs_path", None) - if rootfs is None: - return "-" - path = Path(rootfs) - return path.parent.name or path.name or "-" - - -def vm_ssh_port(vm: Any) -> str: - port = getattr(getattr(vm, "network", None), "ssh_host_port", None) - return str(port) if port is not None else "-" diff --git a/tests/test_build_debian_image.py b/tests/test_build_debian_image.py index e234756..034f32f 100644 --- a/tests/test_build_debian_image.py +++ b/tests/test_build_debian_image.py @@ -77,7 +77,7 @@ def test_packaged_pi_containerfile_uses_npm_global_install() -> None: assert 'export PATH="$HOME/.local/bin:$HOME/.nodejs/bin:$PATH"' in content -def test_build_debian_image_omits_sdk_sketch_by_default( +def test_build_debian_image_prints_summary( fake_smolvm_images: None, tmp_path: Path, capsys: pytest.CaptureFixture[str], @@ -94,134 +94,10 @@ def test_build_debian_image_omits_sdk_sketch_by_default( assert "sbx config:" in out assert "disk_size" not in out assert "run_user = 'agent'" in out - assert "SDK usage sketch:" not in out - assert "from smolvm import SmolVM" not in out manifest = json.loads((tmp_path / "image" / "smolvm-image.json").read_text()) assert manifest["sbx"]["features"] == [] -def test_build_debian_image_prints_sdk_sketch_when_requested( - fake_smolvm_images: None, - tmp_path: Path, - capsys: pytest.CaptureFixture[str], -) -> None: - module = _load_module() - key = tmp_path / "id_ed25519.pub" - key.write_text("ssh-ed25519 fake", encoding="utf-8") - - rc = module.main(["--ssh-public-key", str(key), "--name", "image", "--sdk-sketch"]) - - assert rc == 0 - out = capsys.readouterr().out - assert "SDK usage sketch:" in out - assert "from smolvm import SmolVM, VMConfig" in out - - -def test_build_debian_image_prints_sdk_sketch_for_existing_image_without_rebuild( - tmp_path: Path, - capsys: pytest.CaptureFixture[str], -) -> None: - module = _load_module() - image = tmp_path / "image" - image.mkdir() - (image / "smolvm-image.json").write_text( - '{"kernel":"vmlinux.bin","rootfs":"rootfs.ext4","boot_args":"boot args"}', - encoding="utf-8", - ) - - rc = module.main(["--print-sdk-sketch", str(image)]) - - assert rc == 0 - out = capsys.readouterr().out - assert "SDK usage sketch:" in out - assert f"kernel_path=Path('{image / 'vmlinux.bin'}')" in out - assert f"rootfs_path=Path('{image / 'rootfs.ext4'}')" in out - assert "boot_args='boot args'" in out - - -def test_print_sdk_sketch_reports_missing_manifest( - tmp_path: Path, - capsys: pytest.CaptureFixture[str], -) -> None: - module = _load_module() - image = tmp_path / "image" - image.mkdir() - - rc = module.main(["--print-sdk-sketch", str(image)]) - - assert rc == 2 - assert "manifest not found" in capsys.readouterr().err - - -def test_print_sdk_sketch_reports_invalid_manifest_json( - tmp_path: Path, - capsys: pytest.CaptureFixture[str], -) -> None: - module = _load_module() - image = tmp_path / "image" - image.mkdir() - (image / "smolvm-image.json").write_text("not json", encoding="utf-8") - - rc = module.main(["--print-sdk-sketch", str(image)]) - - assert rc == 2 - assert "invalid image manifest" in capsys.readouterr().err - - -def test_print_sdk_sketch_rejects_non_string_manifest_paths( - tmp_path: Path, - capsys: pytest.CaptureFixture[str], -) -> None: - module = _load_module() - image = tmp_path / "image" - image.mkdir() - (image / "smolvm-image.json").write_text( - '{"kernel":123,"rootfs":"rootfs.ext4"}', - encoding="utf-8", - ) - - rc = module.main(["--print-sdk-sketch", str(image)]) - - assert rc == 2 - assert "kernel and rootfs must be strings" in capsys.readouterr().err - - -def test_print_sdk_sketch_rejects_non_string_boot_args( - tmp_path: Path, - capsys: pytest.CaptureFixture[str], -) -> None: - module = _load_module() - image = tmp_path / "image" - image.mkdir() - (image / "smolvm-image.json").write_text( - '{"kernel":"vmlinux.bin","rootfs":"rootfs.ext4","boot_args":123}', - encoding="utf-8", - ) - - rc = module.main(["--print-sdk-sketch", str(image)]) - - assert rc == 2 - assert "boot_args must be a string" in capsys.readouterr().err - - -def test_print_sdk_sketch_uses_default_boot_args( - tmp_path: Path, - capsys: pytest.CaptureFixture[str], -) -> None: - module = _load_module() - image = tmp_path / "image" - image.mkdir() - (image / "smolvm-image.json").write_text( - '{"kernel":"vmlinux.bin","rootfs":"rootfs.ext4"}', - encoding="utf-8", - ) - - rc = module.main(["--print-sdk-sketch", str(image)]) - - assert rc == 0 - assert "console=ttyS0 reboot=k panic=1" in capsys.readouterr().out - - def test_build_debian_image_missing_ssh_key_returns_2( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, @@ -260,22 +136,6 @@ def test_build_debian_image_rejects_with_docker_and_custom_containerfile( assert "cannot be combined" in capsys.readouterr().err -def test_build_debian_image_import_error_returns_127( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, - capsys: pytest.CaptureFixture[str], -) -> None: - module = _load_module() - key = tmp_path / "id_ed25519.pub" - key.write_text("ssh-ed25519 fake", encoding="utf-8") - monkeypatch.setitem(sys.modules, "smolvm.images.builder", None) - - rc = module.main(["--ssh-public-key", str(key)]) - - assert rc == 127 - assert "smolvm is not installed" in capsys.readouterr().err - - def test_build_debian_image_json_output( fake_smolvm_images: None, tmp_path: Path, diff --git a/tests/test_cli.py b/tests/test_cli.py index b22f14a..de967ac 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -22,6 +22,14 @@ def isolated_config(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: monkeypatch.setattr(guest_setup, "host_git_config", lambda project_root=None: None) monkeypatch.setattr(guest_setup, "set_hostname", lambda *args, **kwargs: None) monkeypatch.setattr(guest_setup, "sync_guest_clock", lambda *args, **kwargs: None) + monkeypatch.setattr(guest_setup, "attach", lambda *args, **kwargs: 0) + monkeypatch.setattr( + cli.smolvm_preset, + "create_preset", + lambda preset_name, **kwargs: SimpleNamespace( + vm_id=kwargs.get("vm_name") or f"{preset_name}-sbx", close=lambda: None + ), + ) def install_fake_smolvm(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path: @@ -41,6 +49,22 @@ def print_smolvm_args(monkeypatch: pytest.MonkeyPatch) -> None: ) +def capture_preset( + monkeypatch: pytest.MonkeyPatch, + captured: dict[str, object], + *, + vm_id: str = "vm1", +) -> None: + def fake_create(preset_name: str, **kwargs: object) -> SimpleNamespace: + captured["preset"] = (preset_name, kwargs) + return SimpleNamespace( + vm_id=vm_id, + close=lambda: captured.setdefault("preset_closed", True), + ) + + monkeypatch.setattr(cli.smolvm_preset, "create_preset", fake_create) + + def test_smolvm_runner_does_not_need_console_script_on_path() -> None: assert runtime.smolvm_argv(["doctor"]) == [ sys.executable, @@ -443,13 +467,18 @@ def test_run_uses_local_toml_defaults( encoding="utf-8", ) monkeypatch.setattr(cli, "LOCAL_CONFIG_PATHS", (local_config,)) + monkeypatch.setattr(cli, "_post_start_actions", lambda **kwargs: 0) + captured: dict[str, object] = {} + capture_preset(monkeypatch, captured, vm_id="demo") rc = cli.main(["run", "--no-auth-port"]) assert rc == 0 - assert capfd.readouterr().out == ( - "codex start --name demo --backend qemu --boot-timeout 60 --mount .:/workspace --attach\n" - ) + preset_name, preset_kwargs = captured["preset"] + assert preset_name == "codex" + assert preset_kwargs["vm_name"] == "demo" + assert preset_kwargs["mounts"] == [".:/workspace"] + assert capfd.readouterr().out == "Started 'demo'. Launching codex...\n" def test_run_supports_all_exposed_options_from_config( @@ -471,6 +500,7 @@ def test_run_supports_all_exposed_options_from_config( name = "configured" memory = 8192 disk_size = 32768 +cpus = 4 backend = "qemu" os = "ubuntu" mount = ["{extra_mount}", ".:/workspace"] @@ -478,45 +508,36 @@ def test_run_supports_all_exposed_options_from_config( writable_mounts = false install_timeout = 900 boot_timeout = 75 +port_forwards = ["8080:80"] """.strip(), encoding="utf-8", ) - rc = cli.main(["--config", str(config), "run", "--no-auth-port", "--no-attach"]) - - assert rc == 0 - assert capfd.readouterr().out == ( - "codex start --name configured --memory 8192 --disk-size 32768 " - "--os ubuntu --install-timeout 900 --backend qemu --boot-timeout 75 " - f"--mount {project}:{project} " - f"--mount {extra_mount}:{extra_mount} " - "--mount .:/workspace --writable-mounts --no-attach\n" - ) - - -def test_run_with_cpus_uses_sdk_start_path( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - install_fake_smolvm(monkeypatch, tmp_path) - config = tmp_path / "config.toml" - config.write_text( - '[sbx]\nagent = "pi"\ncpus = 4\nauth_port = false\ngit_config = false\n', - encoding="utf-8", - ) captured: dict[str, object] = {} + capture_preset(monkeypatch, captured, vm_id="configured") - def fake_start_preset_with_sdk(**kwargs: object) -> int: - captured.update(kwargs) - return 0 - - monkeypatch.setattr(cli, "_start_preset_with_sdk", fake_start_preset_with_sdk) - - rc = cli.main(["--config", str(config), "run", "--no-attach"]) + rc = cli.main(["--config", str(config), "run", "--no-auth-port", "--no-attach"]) assert rc == 0 - assert captured["agent"] == "pi" - assert captured["cpus"] == 4 + preset_name, preset_kwargs = captured["preset"] + assert preset_name == "codex" + assert preset_kwargs["vm_name"] == "configured" + assert preset_kwargs["memory_mib"] == 8192 + assert preset_kwargs["disk_size_mib"] == 32768 + assert preset_kwargs["cpus"] == 4 + assert preset_kwargs["guest_os"] == "ubuntu" + assert preset_kwargs["install_timeout"] == 900 + assert preset_kwargs["boot_timeout"] == 75 + assert preset_kwargs["mounts"] == [ + f"{project}:{project}", + f"{extra_mount}:{extra_mount}", + ".:/workspace", + ] + assert preset_kwargs["writable_mounts"] is True + assert preset_kwargs["port_forwards"] == [ + {"host_address": "127.0.0.1", "host_port": 8080, "guest_port": 80} + ] + assert capfd.readouterr().out == "Started 'configured'.\n" def test_run_uses_local_image_directory( @@ -550,6 +571,13 @@ def fake_start_local_image(**kwargs: object) -> int: return 0 monkeypatch.setattr(cli, "_start_local_image", fake_start_local_image) + monkeypatch.setattr( + cli.smolvm_preset, + "create_preset", + lambda *args, **kwargs: (_ for _ in ()).throw( + AssertionError("local images must not install presets") + ), + ) rc = cli.main(["--config", str(config), "run", "--no-auth-port"]) @@ -560,28 +588,14 @@ def fake_start_local_image(**kwargs: object) -> int: assert captured["attach"] is True -def test_run_user_from_config_starts_without_smolvm_attach_then_attaches_as_user( +def test_run_user_from_config_starts_then_attaches_as_user( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capfd: pytest.CaptureFixture[str], ) -> None: install_fake_smolvm(monkeypatch, tmp_path) captured: dict[str, object] = {} - - def fake_run_capture( - argv: list[str], *, env: dict[str, str] | None = None - ) -> subprocess.CompletedProcess[str]: - captured["argv"] = argv - captured["env"] = env - return subprocess.CompletedProcess( - argv, - 0, - stdout=( - '{"ok": true, "data": {"vm": {"name": "vm1"}}, ' - '"command": "start", "exit_code": 0, "error": null}\n' - ), - stderr="", - ) + capture_preset(monkeypatch, captured) def fake_prepare(vm_id: str, user: str, **kwargs: object) -> None: captured["prepare"] = (vm_id, user) @@ -590,33 +604,16 @@ def fake_attach(vm_id: str, launch_command: str, **kwargs: object) -> int: captured["attach"] = (vm_id, kwargs.get("user"), launch_command, kwargs.get("cwd")) return 0 - monkeypatch.setattr(cli.runtime, "run_capture", fake_run_capture) monkeypatch.setattr(cli.network, "expose_auth_port", lambda vm_id, host_port, guest_port: 0) monkeypatch.setattr(guest_setup, "prepare_run_user", fake_prepare) monkeypatch.setattr(guest_setup, "attach", fake_attach) config = tmp_path / "config.toml" - config.write_text( - """ -[sbx] -run_user = "agent" -""".strip(), - encoding="utf-8", - ) + config.write_text('[sbx]\nrun_user = "agent"\n', encoding="utf-8") rc = cli.main(["--config", str(config), "run"]) assert rc == 0 - assert captured["argv"] == [ - "smolvm", - "pi", - "start", - "--backend", - "qemu", - "--boot-timeout", - "60", - "--no-attach", - "--json", - ] + assert captured["preset_closed"] is True assert captured["prepare"] == ("vm1", "agent") assert captured["attach"] == ("vm1", "agent", "pi", None) assert capfd.readouterr().out == ( @@ -672,17 +669,7 @@ def test_git_config_defaults_on_for_managed_run( ) monkeypatch.setattr(guest_setup, "attach", lambda *args, **kwargs: 0) - def fake_run_capture( - argv: list[str], *, env: dict[str, str] | None = None - ) -> subprocess.CompletedProcess[str]: - return subprocess.CompletedProcess( - argv, - 0, - stdout='{ "ok": true, "data": {"vm": {"name": "vm1"}} }\n', - stderr="", - ) - - monkeypatch.setattr(cli.runtime, "run_capture", fake_run_capture) + capture_preset(monkeypatch, captured) assert cli.main(["run"]) == 0 assert captured["git"] == ("vm1", None, "[user]\n\tname = Test\n") @@ -705,17 +692,7 @@ def test_no_git_config_disables_forwarding( ) monkeypatch.setattr(guest_setup, "attach", lambda *args, **kwargs: 0) - def fake_run_capture( - argv: list[str], *, env: dict[str, str] | None = None - ) -> subprocess.CompletedProcess[str]: - return subprocess.CompletedProcess( - argv, - 0, - stdout='{ "ok": true, "data": {"vm": {"name": "vm1"}} }\n', - stderr="", - ) - - monkeypatch.setattr(cli.runtime, "run_capture", fake_run_capture) + capture_preset(monkeypatch, captured) assert cli.main(["run", "--no-git-config"]) == 0 assert captured["git"] == ("vm1", None, None) @@ -748,32 +725,18 @@ def fake_credential_free_env(temp_home: Path, *, forward_env: list[str]) -> dict captured["temp_home"] = temp_home return {"HOME": str(temp_home), "SBX_TEST": "credential-free"} - def fake_run(argv: list[str], *, check: bool = False, env: dict[str, str] | None = None) -> int: - captured["argv"] = argv - captured["env"] = env - return 0 - monkeypatch.setattr(guest_setup, "credential_free_env", fake_credential_free_env) - monkeypatch.setattr(cli.runtime, "run", fake_run) + capture_preset(monkeypatch, captured, vm_id="pi-sbx") rc = cli.main(["run", "--no-attach", "--no-auth-port"]) assert rc == 0 - assert captured["argv"] == [ - "smolvm", - "pi", - "start", - "--backend", - "qemu", - "--boot-timeout", - "60", - "--no-attach", - ] - assert captured["env"] == { + _, preset_kwargs = captured["preset"] + assert preset_kwargs["host_env"] == { "HOME": str(captured["temp_home"]), "SBX_TEST": "credential-free", - "SMOLVM_DISABLE_VERSION_CHECK": "1", } + assert not Path(captured["temp_home"]).exists() def test_env_vars_are_not_forwarded_by_default_with_host_credentials( @@ -784,17 +747,13 @@ def test_env_vars_are_not_forwarded_by_default_with_host_credentials( monkeypatch.setenv("OPENAI_API_KEY", "secret") captured: dict[str, object] = {} - def fake_run(argv: list[str], *, check: bool = False, env: dict[str, str] | None = None) -> int: - captured["env"] = env - return 0 - - monkeypatch.setattr(cli.runtime, "run", fake_run) + capture_preset(monkeypatch, captured, vm_id="pi-sbx") rc = cli.main(["run", "--no-attach", "--copy-host-credentials", "--no-auth-port"]) assert rc == 0 - assert captured["env"] is not None - assert "OPENAI_API_KEY" not in captured["env"] + _, preset_kwargs = captured["preset"] + assert "OPENAI_API_KEY" not in preset_kwargs["host_env"] def test_env_flag_explicitly_forwards_selected_env_var( @@ -805,11 +764,7 @@ def test_env_flag_explicitly_forwards_selected_env_var( monkeypatch.setenv("OPENAI_API_KEY", "secret") captured: dict[str, object] = {} - def fake_run(argv: list[str], *, check: bool = False, env: dict[str, str] | None = None) -> int: - captured["env"] = env - return 0 - - monkeypatch.setattr(cli.runtime, "run", fake_run) + capture_preset(monkeypatch, captured, vm_id="pi-sbx") rc = cli.main( [ @@ -823,7 +778,8 @@ def fake_run(argv: list[str], *, check: bool = False, env: dict[str, str] | None ) assert rc == 0 - assert captured["env"]["OPENAI_API_KEY"] == "secret" + _, preset_kwargs = captured["preset"] + assert preset_kwargs["host_env"]["OPENAI_API_KEY"] == "secret" def test_copy_host_credentials_flag_uses_current_environment( @@ -836,28 +792,14 @@ def test_copy_host_credentials_flag_uses_current_environment( def fail_credential_free_env(temp_home: Path, *, forward_env: list[str]) -> dict[str, str]: raise AssertionError("credential-free env should not be created") - def fake_run(argv: list[str], *, check: bool = False, env: dict[str, str] | None = None) -> int: - captured["argv"] = argv - captured["env"] = env - return 0 - monkeypatch.setattr(guest_setup, "credential_free_env", fail_credential_free_env) - monkeypatch.setattr(cli.runtime, "run", fake_run) + capture_preset(monkeypatch, captured, vm_id="pi-sbx") rc = cli.main(["run", "--no-attach", "--copy-host-credentials", "--no-auth-port"]) assert rc == 0 - assert captured["argv"] == [ - "smolvm", - "pi", - "start", - "--backend", - "qemu", - "--boot-timeout", - "60", - "--no-attach", - ] - assert captured["env"] is not None + _, preset_kwargs = captured["preset"] + assert preset_kwargs["host_env"]["HOME"] == os.environ["HOME"] def test_destroy_deletes_vm( @@ -1003,26 +945,12 @@ def test_run_with_project_path_attaches_from_mounted_project_cwd( project = tmp_path / "project" project.mkdir() captured: dict[str, object] = {} - - def fake_run_capture( - argv: list[str], *, env: dict[str, str] | None = None - ) -> subprocess.CompletedProcess[str]: - captured["argv"] = argv - return subprocess.CompletedProcess( - argv, - 0, - stdout=( - '{"ok": true, "data": {"vm": {"name": "vm1"}}, ' - '"command": "start", "exit_code": 0, "error": null}\n' - ), - stderr="", - ) + capture_preset(monkeypatch, captured) def fake_attach(vm_id: str, launch_command: str, **kwargs: object) -> int: captured["attach"] = (vm_id, launch_command, kwargs.get("cwd")) return 0 - monkeypatch.setattr(cli.runtime, "run_capture", fake_run_capture) monkeypatch.setattr(guest_setup, "attach", fake_attach) rc = cli.main( @@ -1036,20 +964,9 @@ def fake_attach(vm_id: str, launch_command: str, **kwargs: object) -> int: ) assert rc == 0 - assert captured["argv"] == [ - "smolvm", - "pi", - "start", - "--backend", - "qemu", - "--boot-timeout", - "60", - "--mount", - f"{project}:{project}", - "--writable-mounts", - "--no-attach", - "--json", - ] + _, preset_kwargs = captured["preset"] + assert preset_kwargs["mounts"] == [f"{project}:{project}"] + assert preset_kwargs["writable_mounts"] is True assert captured["attach"] == ("vm1", "pi", str(project)) @@ -1059,20 +976,7 @@ def test_run_exposes_auth_port_by_default_before_attach( ) -> None: install_fake_smolvm(monkeypatch, tmp_path) captured: dict[str, object] = {} - - def fake_run_capture( - argv: list[str], *, env: dict[str, str] | None = None - ) -> subprocess.CompletedProcess[str]: - captured["argv"] = argv - return subprocess.CompletedProcess( - argv, - 0, - stdout=( - '{"ok": true, "data": {"vm": {"name": "vm1"}}, ' - '"command": "start", "exit_code": 0, "error": null}\n' - ), - stderr="", - ) + capture_preset(monkeypatch, captured) def fake_expose(vm_id: str, host_port: int, guest_port: int) -> int: captured["expose"] = (vm_id, host_port, guest_port) @@ -1082,24 +986,11 @@ def fake_attach(vm_id: str, launch_command: str, **kwargs: object) -> int: captured["attach"] = (vm_id, launch_command, kwargs.get("cwd")) return 0 - monkeypatch.setattr(cli.runtime, "run_capture", fake_run_capture) monkeypatch.setattr(cli.network, "expose_auth_port", fake_expose) monkeypatch.setattr(guest_setup, "attach", fake_attach) - rc = cli.main(["run", "--copy-host-credentials"]) - - assert rc == 0 - assert captured["argv"] == [ - "smolvm", - "pi", - "start", - "--backend", - "qemu", - "--boot-timeout", - "60", - "--no-attach", - "--json", - ] + assert cli.main(["run", "--copy-host-credentials"]) == 0 + assert captured["preset_closed"] is True assert captured["expose"] == ("vm1", 1455, 1455) assert captured["attach"] == ("vm1", "pi", None) @@ -1131,6 +1022,13 @@ def fake_run(argv: list[str], *, check: bool = False, env: dict[str, str] | None monkeypatch.setattr(cli.runtime, "run_capture", fake_run_capture) monkeypatch.setattr(cli.runtime, "run", fake_run) + monkeypatch.setattr( + cli.smolvm_preset, + "create_preset", + lambda *args, **kwargs: (_ for _ in ()).throw( + AssertionError("existing VMs must not install presets") + ), + ) monkeypatch.setattr(guest_setup, "sync_guest_clock", lambda vm_id, **kwargs: None) monkeypatch.setattr(cli.network, "expose_auth_port", lambda vm_id, host_port, guest_port: 0) monkeypatch.setattr(guest_setup, "attach", lambda *args, **kwargs: 0) @@ -1180,77 +1078,49 @@ def test_failed_managed_run_hides_json_and_prints_hint( ) -> None: install_fake_smolvm(monkeypatch, tmp_path) - def fake_run_capture( - argv: list[str], *, env: dict[str, str] | None = None - ) -> subprocess.CompletedProcess[str]: - if argv[:2] == ["smolvm", "sandbox", "info"]: - return subprocess.CompletedProcess(argv, 1, stdout="", stderr="not found") - return subprocess.CompletedProcess( - argv, - 1, - stdout=( - '{"ok": false, "error": {"message": ' - "\"QEMU exited early while booting VM '999714'\"}}\n" - ), - stderr="", - ) - - monkeypatch.setattr(cli.runtime, "run_capture", fake_run_capture) + monkeypatch.setattr( + cli.smolvm_preset, + "create_preset", + lambda *args, **kwargs: (_ for _ in ()).throw( + RuntimeError("QEMU exited early while booting VM 'vm1'") + ), + ) rc = cli.main(["run", "--name", "vm1"]) output = capsys.readouterr() assert rc == 1 assert output.out == "" + assert "failed to create preset 'pi'" in output.err assert "QEMU exited early" in output.err - assert "sbx recreate --force" in output.err -def test_run_positional_name_before_options_does_not_pass_sbx_flags_to_smolvm( +@pytest.mark.parametrize( + ("error", "expected_rc"), + [(RuntimeError("failed"), 1), (KeyboardInterrupt(), 130)], +) +def test_preset_failure_cleans_temporary_home( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, + error: BaseException, + expected_rc: int, ) -> None: install_fake_smolvm(monkeypatch, tmp_path) - captured: dict[str, object] = {} - - def fake_run_capture( - argv: list[str], *, env: dict[str, str] | None = None - ) -> subprocess.CompletedProcess[str]: - if argv[:2] == ["smolvm", "sandbox", "info"]: - return subprocess.CompletedProcess(argv, 1, stdout="", stderr="not found") - captured["create"] = argv - return subprocess.CompletedProcess( - argv, - 0, - stdout=( - '{"ok": true, "data": {"vm": {"name": "pi-sbx"}}, ' - '"command": "start", "exit_code": 0, "error": null}\n' - ), - stderr="", - ) + captured: dict[str, Path] = {} - def fake_run(argv: list[str], *, check: bool = False, env: dict[str, str] | None = None) -> int: - captured["create"] = argv - return 0 + def fake_credential_free_env(temp_home: Path, *, forward_env: list[str]) -> dict[str, str]: + captured["temp_home"] = temp_home + return {"HOME": str(temp_home)} - monkeypatch.setattr(cli.runtime, "run_capture", fake_run_capture) - monkeypatch.setattr(cli.runtime, "run", fake_run) + monkeypatch.setattr(guest_setup, "credential_free_env", fake_credential_free_env) + monkeypatch.setattr( + cli.smolvm_preset, + "create_preset", + lambda *args, **kwargs: (_ for _ in ()).throw(error), + ) - rc = cli.main(["run", "pi-sbx", "--no-auth-port", "--no-attach"]) - - assert rc == 0 - assert captured["create"] == [ - "smolvm", - "pi", - "start", - "--name", - "pi-sbx", - "--backend", - "qemu", - "--boot-timeout", - "60", - "--no-attach", - ] + assert cli.main(["create", "vm1", "--no-auth-port"]) == expected_rc + assert not captured["temp_home"].exists() def test_run_positional_name_creates_missing_vm( @@ -1259,43 +1129,14 @@ def test_run_positional_name_creates_missing_vm( ) -> None: install_fake_smolvm(monkeypatch, tmp_path) captured: dict[str, object] = {} + capture_preset(monkeypatch, captured, vm_id="pi-sbx") + monkeypatch.setattr(cli.network, "expose_auth_port", lambda *args: 0) - def fake_run_capture( - argv: list[str], *, env: dict[str, str] | None = None - ) -> subprocess.CompletedProcess[str]: - if argv[:2] == ["smolvm", "sandbox", "info"]: - return subprocess.CompletedProcess(argv, 1, stdout="", stderr="not found") - captured["create"] = argv - return subprocess.CompletedProcess( - argv, - 0, - stdout=( - '{"ok": true, "data": {"vm": {"name": "pi-sbx"}}, ' - '"command": "start", "exit_code": 0, "error": null}\n' - ), - stderr="", - ) - - monkeypatch.setattr(cli.runtime, "run_capture", fake_run_capture) - monkeypatch.setattr(cli.network, "expose_auth_port", lambda vm_id, host_port, guest_port: 0) - monkeypatch.setattr(guest_setup, "attach", lambda *args, **kwargs: 0) - - rc = cli.main(["run", "pi-sbx"]) - - assert rc == 0 - assert captured["create"] == [ - "smolvm", - "pi", - "start", - "--name", - "pi-sbx", - "--backend", - "qemu", - "--boot-timeout", - "60", - "--no-attach", - "--json", - ] + assert cli.main(["run", "pi-sbx"]) == 0 + preset_name, preset_kwargs = captured["preset"] + assert preset_name == "pi" + assert preset_kwargs["vm_name"] == "pi-sbx" + assert captured["preset_closed"] is True def test_run_missing_vm_creates_it( @@ -1304,43 +1145,13 @@ def test_run_missing_vm_creates_it( ) -> None: install_fake_smolvm(monkeypatch, tmp_path) captured: dict[str, object] = {} + capture_preset(monkeypatch, captured) + monkeypatch.setattr(cli.network, "expose_auth_port", lambda *args: 0) - def fake_run_capture( - argv: list[str], *, env: dict[str, str] | None = None - ) -> subprocess.CompletedProcess[str]: - if argv[:2] == ["smolvm", "sandbox", "info"]: - return subprocess.CompletedProcess(argv, 1, stdout="", stderr="not found") - captured["create"] = argv - return subprocess.CompletedProcess( - argv, - 0, - stdout=( - '{"ok": true, "data": {"vm": {"name": "vm1"}}, ' - '"command": "start", "exit_code": 0, "error": null}\n' - ), - stderr="", - ) - - monkeypatch.setattr(cli.runtime, "run_capture", fake_run_capture) - monkeypatch.setattr(cli.network, "expose_auth_port", lambda vm_id, host_port, guest_port: 0) - monkeypatch.setattr(guest_setup, "attach", lambda *args, **kwargs: 0) - - rc = cli.main(["run", "--name", "vm1", "--copy-host-credentials"]) - - assert rc == 0 - assert captured["create"] == [ - "smolvm", - "pi", - "start", - "--name", - "vm1", - "--backend", - "qemu", - "--boot-timeout", - "60", - "--no-attach", - "--json", - ] + assert cli.main(["run", "--name", "vm1", "--copy-host-credentials"]) == 0 + _, preset_kwargs = captured["preset"] + assert preset_kwargs["vm_name"] == "vm1" + assert preset_kwargs["host_env"]["HOME"] == os.environ["HOME"] def test_create_is_run_no_attach_alias( @@ -1349,28 +1160,33 @@ def test_create_is_run_no_attach_alias( ) -> None: install_fake_smolvm(monkeypatch, tmp_path) captured: dict[str, object] = {} + capture_preset(monkeypatch, captured) + monkeypatch.setattr( + cli, + "_post_start_actions", + lambda **kwargs: captured.setdefault("post", kwargs) and 0, + ) - def fake_run(argv: list[str], *, check: bool = False, env: dict[str, str] | None = None) -> int: - captured["argv"] = argv - return 0 + assert ( + cli.main(["create", "--name", "vm1", "--copy-host-credentials", "--no-auth-port"]) + == 0 + ) + assert captured["post"]["attach"] is False - monkeypatch.setattr(cli.runtime, "run", fake_run) - rc = cli.main(["create", "--name", "vm1", "--copy-host-credentials", "--no-auth-port"]) +def test_preset_creation_json_is_sbx_owned( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + install_fake_smolvm(monkeypatch, tmp_path) + captured: dict[str, object] = {} + capture_preset(monkeypatch, captured) - assert rc == 0 - assert captured["argv"] == [ - "smolvm", - "pi", - "start", - "--name", - "vm1", - "--backend", - "qemu", - "--boot-timeout", - "60", - "--no-attach", - ] + assert cli.main(["create", "vm1", "--json", "--no-auth-port"]) == 0 + assert json.loads(capsys.readouterr().out) == { + "vm": {"name": "vm1", "status": "running"} + } def test_expose_auth_port_warns_when_port_already_listening( @@ -1618,18 +1434,15 @@ def test_recreate_deletes_then_starts_vm( tmp_path: Path, ) -> None: install_fake_smolvm(monkeypatch, tmp_path) - calls: list[list[str] | tuple[str, str]] = [] + calls: list[tuple[str, str]] = [] + captured: dict[str, object] = {} def fake_delete(vm_id: str, extra_args: list[str] | None = None) -> int: calls.append(("delete", vm_id)) return 0 - def fake_run(argv: list[str], *, check: bool = False, env: dict[str, str] | None = None) -> int: - calls.append(argv) - return 0 - monkeypatch.setattr(cli, "_delete_vm", fake_delete) - monkeypatch.setattr(cli.runtime, "run", fake_run) + capture_preset(monkeypatch, captured) rc = cli.main( [ @@ -1644,21 +1457,9 @@ def fake_run(argv: list[str], *, check: bool = False, env: dict[str, str] | None ) assert rc == 0 - assert calls == [ - ("delete", "vm1"), - [ - "smolvm", - "pi", - "start", - "--name", - "vm1", - "--backend", - "qemu", - "--boot-timeout", - "60", - "--no-attach", - ], - ] + assert calls == [("delete", "vm1")] + _, preset_kwargs = captured["preset"] + assert preset_kwargs["vm_name"] == "vm1" def test_cli_project_path_overrides_config( @@ -1680,6 +1481,9 @@ def test_cli_project_path_overrides_config( encoding="utf-8", ) + captured: dict[str, object] = {} + capture_preset(monkeypatch, captured, vm_id="pi-sbx") + rc = cli.main( [ "--config", @@ -1693,11 +1497,10 @@ def test_cli_project_path_overrides_config( ) assert rc == 0 - assert capfd.readouterr().out == ( - "pi start --backend qemu --boot-timeout 60 " - f"--mount {cli_project}:{cli_project} " - "--writable-mounts --no-attach\n" - ) + _, preset_kwargs = captured["preset"] + assert preset_kwargs["mounts"] == [f"{cli_project}:{cli_project}"] + assert preset_kwargs["writable_mounts"] is True + assert capfd.readouterr().out == "Started 'pi-sbx'.\n" def test_cli_flags_override_config( @@ -1716,6 +1519,10 @@ def test_cli_flags_override_config( encoding="utf-8", ) + captured: dict[str, object] = {} + capture_preset(monkeypatch, captured, vm_id="from-cli") + monkeypatch.setattr(cli, "_post_start_actions", lambda **kwargs: 0) + rc = cli.main( [ "--config", @@ -1730,9 +1537,10 @@ def test_cli_flags_override_config( ) assert rc == 0 - assert capfd.readouterr().out == ( - "claude start --name from-cli --backend qemu --boot-timeout 60 --attach\n" - ) + preset_name, preset_kwargs = captured["preset"] + assert preset_name == "claude" + assert preset_kwargs["vm_name"] == "from-cli" + assert capfd.readouterr().out == "Started 'from-cli'. Launching claude...\n" def test_invalid_agent_in_config_returns_usage_error( diff --git a/tests/test_cli_extra.py b/tests/test_cli_extra.py index b77d75b..114e927 100644 --- a/tests/test_cli_extra.py +++ b/tests/test_cli_extra.py @@ -27,6 +27,13 @@ def isolated_state(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: monkeypatch.setattr(cli, "SMOLVM_DB_PATH", tmp_path / "smolvm.db") monkeypatch.setattr(vm_state, "SMOLVM_DB_PATH", tmp_path / "smolvm.db") monkeypatch.setattr(guest_setup, "set_hostname", lambda *args, **kwargs: None) + monkeypatch.setattr( + cli.smolvm_preset, + "create_preset", + lambda preset_name, **kwargs: SimpleNamespace( + vm_id=kwargs.get("vm_name") or f"{preset_name}-sbx", close=lambda: None + ), + ) @pytest.fixture @@ -437,7 +444,7 @@ def called_process_error(*args: object, **kwargs: object) -> subprocess.Complete assert runtime.run(["cmd"]) == 9 -def test_simple_helper_error_branches(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: +def test_simple_helper_error_branches(tmp_path: Path) -> None: assert cli._deep_merge({"a": {"b": 1}}, {"a": {"c": 2}}) == {"a": {"b": 1, "c": 2}} assert cli._resolve_project_path(str(tmp_path / "missing")) == tmp_path / "missing" assert cli._same_path_mount(str(tmp_path)) == f"{tmp_path}:{tmp_path}" @@ -445,8 +452,6 @@ def test_simple_helper_error_branches(tmp_path: Path, capsys: pytest.CaptureFixt guest_setup.validate_run_user("bad user") with pytest.raises(cli.ConfigError, match="invalid env var"): guest_setup.validate_env_names(["1BAD"]) - cli._print_start_failure('{"error":{"message":"QEMU exited early"}}') - assert "Try `sbx recreate" in capsys.readouterr().err def test_ssh_command_and_pid_alive(monkeypatch: pytest.MonkeyPatch) -> None: @@ -1119,10 +1124,6 @@ def test_config_and_validation_error_branches(tmp_path: Path) -> None: guest_setup.validate_run_user("bad user") with pytest.raises(cli.ConfigError, match="invalid env var"): guest_setup.validate_env_names(["BAD-NAME"]) - with pytest.raises(cli.ConfigError, match="could not read"): - cli._extract_started_vm_name("not-json") - with pytest.raises(cli.ConfigError, match="did not include"): - cli._extract_started_vm_name('{"data":{"vm":{"name":""}}}') def test_command_edge_cases( diff --git a/tests/test_smolvm_preset.py b/tests/test_smolvm_preset.py new file mode 100644 index 0000000..c26b7e6 --- /dev/null +++ b/tests/test_smolvm_preset.py @@ -0,0 +1,208 @@ +import os +from types import SimpleNamespace + +import pytest +from smolvm.types import PortForwardConfig + +from sbx import smolvm_preset + + +class FakeConfig: + def __init__(self, events: list[str]) -> None: + self.events = events + self.updates: dict[str, object] = {} + + def model_copy(self, *, update: dict[str, object]) -> "FakeConfig": + self.events.append("update") + self.updates = update + return self + + +class FakeVM: + vm_id = "demo" + + def __init__(self, events: list[str]) -> None: + self.events = events + + def start(self, *, boot_timeout: float) -> None: + self.events.append(f"start:{boot_timeout:g}") + + def wait_for_ssh(self, *, timeout: float) -> None: + self.events.append(f"wait:{timeout:g}") + + def _ensure_ssh_for_env(self) -> str: + self.events.append("channel") + return "channel" + + def close(self) -> None: + self.events.append("close") + + +def create_preset(**overrides: object) -> object: + options = { + "preset_name": "pi", + "vm_name": None, + "guest_os": "ubuntu", + "cpus": None, + "memory_mib": None, + "disk_size_mib": None, + "mounts": [], + "writable_mounts": False, + "port_forwards": [], + "boot_timeout": 60, + "install_timeout": 600, + "host_env": {"HOME": "/isolated"}, + } + return smolvm_preset.create_preset(**(options | overrides)) + + +@pytest.mark.parametrize( + ("memory", "disk", "cpus", "expected_memory", "expected_disk"), + [(None, None, None, 2048, 8192), (4096, 16384, 4, 4096, 16384)], +) +def test_create_preset( + monkeypatch: pytest.MonkeyPatch, + memory: int | None, + disk: int | None, + cpus: int | None, + expected_memory: int, + expected_disk: int, +) -> None: + events: list[str] = [] + captured: dict[str, object] = {} + config = FakeConfig(events) + vm = FakeVM(events) + monkeypatch.setenv("ORIGINAL", "kept") + + def get_preset(name: str) -> SimpleNamespace: + captured["preset_name"] = name + return SimpleNamespace(name=name, default_mem_mib=2048, default_disk_mib=8192) + + def build_auto_config(**kwargs: object) -> tuple[FakeConfig, str]: + events.append("build") + captured["build"] = kwargs + captured["build_env"] = dict(os.environ) + return config, "private-key" + + def construct_vm(*args: object, **kwargs: object) -> FakeVM: + events.append("construct") + captured["construct"] = (args, kwargs) + return vm + + def apply_preset(channel: object, preset: object, *, install_timeout: int) -> None: + events.append(f"apply:{install_timeout}") + captured["apply"] = (channel, preset) + captured["apply_env"] = dict(os.environ) + + monkeypatch.setattr(smolvm_preset, "get_preset", get_preset) + monkeypatch.setattr(smolvm_preset, "_build_auto_config", build_auto_config) + monkeypatch.setattr(smolvm_preset, "SmolVM", construct_vm) + monkeypatch.setattr(smolvm_preset, "apply_preset", apply_preset) + + result = create_preset( + preset_name="claude", + vm_name="demo", + cpus=cpus, + memory_mib=memory, + disk_size_mib=disk, + mounts=["/host:/guest"], + writable_mounts=True, + port_forwards=[ + {"host_address": "0.0.0.0", "host_port": 3000, "guest_port": 30} + ], + boot_timeout=75, + install_timeout=900, + host_env={"HOME": "/isolated", "ALLOWED_SECRET": "secret"}, + ) + + assert result is vm + assert captured["preset_name"] == "claude-code" + assert captured["build"] == { + "vm_name": "demo", + "name_prefix": "claude", + "os": "ubuntu", + "backend": "qemu", + "memory": expected_memory, + "disk_size_mib": expected_disk, + "ssh_key_path": None, + } + assert captured["build_env"] == {"HOME": "/isolated", "ALLOWED_SECRET": "secret"} + assert captured["apply_env"] == {"HOME": "/isolated", "ALLOWED_SECRET": "secret"} + assert os.environ["ORIGINAL"] == "kept" + assert "ALLOWED_SECRET" not in os.environ + assert events == ["build", "update", "construct", "start:75", "wait:75", "channel", "apply:900"] + assert config.updates.get("vcpu_count") == cpus + if cpus is None: + assert "vcpu_count" not in config.updates + forwards = config.updates["port_forwards"] + assert isinstance(forwards, list) + assert forwards[0] == PortForwardConfig( + host_address="0.0.0.0", host_port=3000, guest_port=30 + ) + assert captured["construct"] == ( + (config,), + { + "ssh_key_path": "private-key", + "mounts": ["/host:/guest"], + "writable_mounts": True, + }, + ) + + +@pytest.mark.parametrize("error", [RuntimeError("failed"), KeyboardInterrupt()]) +def test_create_preset_closes_and_restores_environment_on_failure( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + error: BaseException, +) -> None: + events: list[str] = [] + vm = FakeVM(events) + monkeypatch.setenv("ORIGINAL", "kept") + monkeypatch.setattr( + smolvm_preset, + "get_preset", + lambda name: SimpleNamespace(name=name, default_mem_mib=1, default_disk_mib=1), + ) + monkeypatch.setattr( + smolvm_preset, + "_build_auto_config", + lambda **kwargs: (FakeConfig(events), "key"), + ) + monkeypatch.setattr(smolvm_preset, "SmolVM", lambda *args, **kwargs: vm) + monkeypatch.setattr( + smolvm_preset, + "apply_preset", + lambda *args, **kwargs: (_ for _ in ()).throw(error), + ) + + with pytest.raises(type(error)): + create_preset(host_env={"HOME": "/isolated", "SECRET": "hidden"}) + + assert events[-1] == "close" + assert events.count("close") == 1 + assert os.environ["ORIGINAL"] == "kept" + assert "SECRET" not in os.environ + output = capsys.readouterr() + assert "hidden" not in output.out + assert "hidden" not in output.err + + +def test_create_preset_restores_environment_before_vm_construction( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("ORIGINAL", "kept") + monkeypatch.setattr( + smolvm_preset, + "get_preset", + lambda name: SimpleNamespace(name=name, default_mem_mib=1, default_disk_mib=1), + ) + monkeypatch.setattr( + smolvm_preset, + "_build_auto_config", + lambda **kwargs: (_ for _ in ()).throw(RuntimeError("build failed")), + ) + + with pytest.raises(RuntimeError, match="build failed"): + create_preset() + + assert os.environ["ORIGINAL"] == "kept"