Thales Maciel
6e16e74fd5
Fix the default one-shot install path so empty bundled profile directories no longer win over OCI-backed environment pulls or leave broken cached symlinks behind. Treat cached installs as valid only when the manifest and boot artifacts are all present, repair invalid installs on the next pull, and add human-mode phase markers for env pull and run without changing JSON output. Align the Python lifecycle example and public docs with the current exec_vm/vm_exec auto-clean semantics, and validate the slice with focused pytest coverage, make check, make dist-check, and a real default-path pull/inspect/run smoke.
53 lines
1.8 KiB
Python
53 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
from pathlib import Path
|
|
from types import ModuleType
|
|
from typing import Any, cast
|
|
|
|
import pytest
|
|
|
|
|
|
def _load_python_lifecycle_module() -> ModuleType:
|
|
path = Path("examples/python_lifecycle.py")
|
|
spec = importlib.util.spec_from_file_location("python_lifecycle", path)
|
|
if spec is None or spec.loader is None:
|
|
raise AssertionError("failed to load python_lifecycle example")
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_python_lifecycle_example_does_not_delete_after_exec(
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
module = _load_python_lifecycle_module()
|
|
calls: list[str] = []
|
|
|
|
class StubPyro:
|
|
def create_vm(self, **kwargs: object) -> dict[str, object]:
|
|
assert kwargs["environment"] == "debian:12"
|
|
calls.append("create_vm")
|
|
return {"vm_id": "vm-123"}
|
|
|
|
def start_vm(self, vm_id: str) -> dict[str, object]:
|
|
assert vm_id == "vm-123"
|
|
calls.append("start_vm")
|
|
return {"vm_id": vm_id, "state": "started"}
|
|
|
|
def exec_vm(self, vm_id: str, *, command: str, timeout_seconds: int) -> dict[str, object]:
|
|
assert vm_id == "vm-123"
|
|
assert command == "git --version"
|
|
assert timeout_seconds == 30
|
|
calls.append("exec_vm")
|
|
return {"vm_id": vm_id, "stdout": "git version 2.43.0\n"}
|
|
|
|
def delete_vm(self, vm_id: str) -> dict[str, object]:
|
|
raise AssertionError(f"unexpected delete_vm({vm_id}) call")
|
|
|
|
cast(Any, module).Pyro = StubPyro
|
|
module.main()
|
|
|
|
assert calls == ["create_vm", "start_vm", "exec_vm"]
|
|
captured = capsys.readouterr()
|
|
assert "git version 2.43.0" in captured.out
|