Skip to main content

​
Environments

An Environment represents a single VM within a session. Each environment runs a simulator and can execute commands, track state, and be snapshotted.

​
Accessing Environments

Environments are accessed through a session: 
from plato.v2 import AsyncPlato, Env

plato = AsyncPlato()
session = await plato.sessions.create(
    envs=[
        Env.simulator("espocrm", alias="crm"),
        Env.simulator("gitea", alias="git"),
    ]
)

# List all environments
for env in session.envs:
    print(f"{env.alias}: {env.job_id}")

# Get specific environment by alias
crm = session.get_env("crm")

​
Environment Operations

​
Execute Commands

Run shell commands on the environment: 
# Basic command
result = await env.execute("ls -la /app")
print(result.stdout)
print(result.stderr)
print(result.exit_code)

# With timeout
result = await env.execute("long-running-script.sh", timeout=120)

​
Reset

Reset the environment to its initial state: 
await env.reset()

​
Get State

Retrieve the current application state: 
state = await env.get_state()

# Access state data
print(state.mutations)  # Database/file changes
print(state.state)      # Current state snapshot

​
Set Date

Change the system date (useful for testing time-dependent features): 
from datetime import datetime

await env.set_date(datetime(2024, 12, 25, 10, 0, 0))

​
Snapshot

Create a checkpoint of the environment: 
result = await env.snapshot()
print(f"Artifact ID: {result.artifact_id}")

​
Get Connection Info

Get network routing information for the VM: 
info = await env.get_connection_info()
print(f"Gateway IP: {info.vm_gateway_ip}")
print(f"Private IP: {info.vm_private_ip}")
print(f"Ready: {info.ready}")

​
Close

Close the environment (stops the VM): 
await env.close()

​
Environment Properties

PropertyTypeDescription
job_idstrUnique job/VM identifier
aliasstrHuman-readable alias
artifact_idstr | NoneSource artifact ID
session_idstrParent session ID

​
Creating Environments

Use the Env helper to specify environment configurations:

​
From Simulator

from plato.v2 import Env

# Default (latest tag, base dataset)
Env.simulator("espocrm")

# With specific tag
Env.simulator("espocrm:staging")
Env.simulator("espocrm", tag="staging")

# With specific dataset
Env.simulator("espocrm", dataset="blank")

# With custom alias
Env.simulator("espocrm", alias="my-crm")

​
From Artifact

# From a specific snapshot
Env.artifact("artifact-abc123")
Env.artifact("artifact-abc123", alias="restored")

​
From Resource

Create a blank VM with custom resources: 
from plato.v2 import Env, SimConfigCompute

Env.resource(
    "custom-service",
    SimConfigCompute(
        cpus=4,
        memory=8192,  # MB
        disk=20000,   # MB
    ),
    alias="beefy-vm"
)

​
Example: Multi-Environment Workflow

from plato.v2 import AsyncPlato, Env

async def multi_env_example():
    plato = AsyncPlato()

    # Create session with multiple environments
    session = await plato.sessions.create(
        envs=[
            Env.simulator("gitea", alias="git"),
            Env.simulator("espocrm", alias="crm"),
        ]
    )

    try:
        # Get environments
        git = session.get_env("git")
        crm = session.get_env("crm")

        # Reset all to initial state
        await session.reset()

        # Execute commands on specific environments
        git_result = await git.execute("git status")
        crm_result = await crm.execute("php artisan migrate:status")

        # Get state from all environments
        state = await session.get_state()
        for job_id, env_state in state.results.items():
            print(f"{job_id}: {len(env_state.mutations)} mutations")

        # Snapshot everything
        snapshot = await session.snapshot()

    finally:
        await session.close()
        await plato.close()

​
Error Handling

try:
    result = await env.execute("invalid-command")
except Exception as e:
    print(f"Command failed: {e}")

# Check command exit code
result = await env.execute("test -f /nonexistent")
if result.exit_code != 0:
    print(f"File not found: {result.stderr}")