"""Integration tests for Image builder methods across all guest OS types.

Every image builder method is tested on every guest OS that supports it.

Tests auto-skip via skip_if_unsupported() when the required runtime is

not installed on this host.

Run with:

pytest tests/test_image_builder.py -v

pytest tests/test_image_builder.py -v -k linux # only linux tests

"""

from __future__ import annotations

import pytest

from cua import Image, Sandbox, skip_if_unsupported

# ── helpers ──────────────────────────────────────────────────────────────────

async def _run(image: Image, cmd: str) -> str:

"""Ephemeral sandbox: run cmd and return stdout. Asserts success."""

async with Sandbox.ephemeral(image, local=True) as sb:

r = await sb.shell.run(cmd, timeout=120)

assert r.success, f"Command failed (rc={r.returncode}): {r.stderr}"

return r.stdout.strip()

# ═══════════════════════════════════════════════════════════════════════════

# Linux (Docker container)

# ═══════════════════════════════════════════════════════════════════════════

class TestLinuxAptInstall:

def _base(self):

return Image.linux()

def setup_method(self):

skip_if_unsupported(self._base())

async def test_single_package(self):

image = self._base().apt_install("curl")

out = await _run(image, "curl --version")

assert "curl" in out.lower()

async def test_multiple_packages(self):

image = self._base().apt_install("curl", "wget")

out = await _run(image, "curl --version && wget --version")

assert "curl" in out.lower()

async def test_chained_calls(self):

image = self._base().apt_install("curl").apt_install("wget")

out = await _run(image, "curl --version && wget --version")

assert "curl" in out.lower()

class TestLinuxPipInstall:

def _base(self):

return Image.linux().apt_install("python3-pip")

def setup_method(self):

skip_if_unsupported(Image.linux())

async def test_single_package(self):

image = self._base().pip_install("httpx")

out = await _run(image, "python3 -c 'import httpx; print(httpx.__version__)'")

assert out # any version string

async def test_multiple_packages(self):

image = self._base().pip_install("httpx", "rich")

out = await _run(image, "python3 -c 'import httpx, rich; print(\"ok\")'")

assert "ok" in out

class TestLinuxUvInstall:

def setup_method(self):

skip_if_unsupported(Image.linux())

async def test_single_package(self):

image = Image.linux().uv_install("httpx")

out = await _run(image, "uv run python -c 'import httpx; print(\"ok\")'")

assert "ok" in out

async def test_multiple_packages(self):

image = Image.linux().uv_install("httpx", "rich")

out = await _run(image, "uv run python -c 'import httpx, rich; print(\"ok\")'")

assert "ok" in out

class TestLinuxRun:

def setup_method(self):

skip_if_unsupported(Image.linux())

async def test_writes_file(self):

image = Image.linux().run("echo 'hello' > /etc/hello.txt")

out = await _run(image, "cat /etc/hello.txt")

assert "hello" in out

async def test_creates_directory(self):

image = Image.linux().run("mkdir -p /opt/myapp")

out = await _run(image, "test -d /opt/myapp && echo exists")

assert "exists" in out

async def test_chained(self):

image = Image.linux().run("mkdir -p /opt/myapp").run("echo 'v1' > /opt/myapp/version.txt")

out = await _run(image, "cat /opt/myapp/version.txt")

assert "v1" in out

class TestLinuxEnv:

def setup_method(self):

skip_if_unsupported(Image.linux())

async def test_single_variable(self):

image = Image.linux().env(MY_VAR="hello_integration")

out = await _run(image, "echo $MY_VAR")

assert "hello_integration" in out

async def test_multiple_variables(self):

image = Image.linux().env(VAR_A="alpha", VAR_B="beta")

out = await _run(image, "echo $VAR_A $VAR_B")

assert "alpha" in out

assert "beta" in out

async def test_variable_used_in_run(self):

image = Image.linux().env(APP_PORT="8080").run("echo port=$APP_PORT > /tmp/port.txt")

out = await _run(image, "cat /tmp/port.txt")

assert "8080" in out

class TestLinuxCopy:

def setup_method(self):

skip_if_unsupported(Image.linux())

async def test_copy_file(self, tmp_path):

src = tmp_path / "hello.txt"

src.write_text("from host")

image = Image.linux().copy(str(src), "/tmp/hello.txt")

out = await _run(image, "cat /tmp/hello.txt")

assert "from host" in out

async def test_copy_preserves_content(self, tmp_path):

src = tmp_path / "data.json"

src.write_text('{"key": "value"}')

image = Image.linux().copy(str(src), "/tmp/data.json")

out = await _run(image, "cat /tmp/data.json")

assert '"key"' in out

class TestLinuxExpose:

def setup_method(self):

skip_if_unsupported(Image.linux())

async def test_expose_port_reachable(self):

"""expose() should not raise — container starts with the port mapped."""

image = Image.linux().expose(9999)

async with Sandbox.ephemeral(image, local=True) as sb:

# Just verify the sandbox is reachable; port mapping is a docker-run concern

r = await sb.shell.run("echo ok")

assert r.success

class TestLinuxFromRegistry:

def setup_method(self):

skip_if_unsupported(Image.linux())

async def test_ubuntu_22(self):

# from_registry with plain images only works if the image has computer-server.

# Use the cua base image which always has it.

image = Image.from_registry("docker.io/trycua/cua-xfce:latest")

out = await _run(image, "cat /etc/os-release")

assert "ubuntu" in out.lower() or "linux" in out.lower()

async def test_debian(self):

# Same constraint — use cua base image.

image = Image.from_registry("docker.io/trycua/cua-xfce:latest")

out = await _run(image, "uname -s")

assert "linux" in out.lower()

class TestLinuxFullChain:

def setup_method(self):

skip_if_unsupported(Image.linux())

async def test_full_chain(self, tmp_path):

src = tmp_path / "app.sh"

src.write_text("#!/bin/sh\necho app-$VERSION\n")

image = (

Image.linux()

.apt_install("curl")

.env(VERSION="1.0.0")

.copy(str(src), "/usr/local/bin/app.sh")

.run("chmod +x /usr/local/bin/app.sh")

.expose(8080)

)

out = await _run(image, "/usr/local/bin/app.sh")

assert "app-1.0.0" in out

# ═══════════════════════════════════════════════════════════════════════════

# macOS VM

# ═══════════════════════════════════════════════════════════════════════════

class TestMacOSBrewInstall:

def setup_method(self):

skip_if_unsupported(Image.macos())

async def test_single_package(self):

image = Image.macos().brew_install("jq")

out = await _run(image, "jq --version")

assert "jq" in out

async def test_multiple_packages(self):

image = Image.macos().brew_install("jq", "wget")

out = await _run(image, "jq --version && wget --version")

assert "jq" in out

class TestMacOSRun:

def setup_method(self):

skip_if_unsupported(Image.macos())

async def test_run_command(self):

image = Image.macos().run("echo setup > /tmp/setup.txt")

out = await _run(image, "cat /tmp/setup.txt")

assert "setup" in out

class TestMacOSEnv:

def setup_method(self):

skip_if_unsupported(Image.macos())

async def test_env_variable(self):

image = Image.macos().env(MY_ENV="macos_test")

out = await _run(image, "echo $MY_ENV")

assert "macos_test" in out

class TestMacOSCopy:

def setup_method(self):

skip_if_unsupported(Image.macos())

async def test_copy_file(self, tmp_path):

src = tmp_path / "hello.txt"

src.write_text("hello from host")

image = Image.macos().copy(str(src), "/tmp/hello.txt")

out = await _run(image, "cat /tmp/hello.txt")

assert "hello from host" in out

class TestMacOSPipInstall:

def setup_method(self):

skip_if_unsupported(Image.macos())

async def test_pip_install(self):

image = Image.macos().pip_install("httpx")

out = await _run(image, "python3 -c 'import httpx; print(\"ok\")'")

assert "ok" in out

# ═══════════════════════════════════════════════════════════════════════════

# Windows VM

# ═══════════════════════════════════════════════════════════════════════════

class TestWindowsChocoInstall:

def setup_method(self):

skip_if_unsupported(Image.windows())

async def test_single_package(self):

image = Image.windows().choco_install("curl")

out = await _run(image, "curl --version")

assert "curl" in out.lower()

async def test_multiple_packages(self):

image = Image.windows().choco_install("curl", "wget")

out = await _run(image, "curl --version")

assert "curl" in out.lower()

class TestWindowsWingetInstall:

def setup_method(self):

skip_if_unsupported(Image.windows())

async def test_single_package(self):

image = Image.windows().winget_install("Microsoft.PowerShell")

out = await _run(image, "pwsh --version")

assert "powershell" in out.lower()

class TestWindowsRun:

def setup_method(self):

skip_if_unsupported(Image.windows())

async def test_run_powershell(self):

image = Image.windows().run("echo setup > C:\\setup.txt")

out = await _run(image, "type C:\\setup.txt")

assert "setup" in out

class TestWindowsEnv:

def setup_method(self):

skip_if_unsupported(Image.windows())

async def test_env_variable(self):

image = Image.windows().env(MY_VAR="win_test")

out = await _run(image, "echo %MY_VAR%")

assert "win_test" in out

class TestWindowsCopy:

def setup_method(self):

skip_if_unsupported(Image.windows())

async def test_copy_file(self, tmp_path):

src = tmp_path / "hello.txt"

src.write_text("hello windows")

image = Image.windows().copy(str(src), "C:\\hello.txt")

out = await _run(image, "type C:\\hello.txt")

assert "hello windows" in out

class TestWindowsExpose:

def setup_method(self):

skip_if_unsupported(Image.windows())

async def test_expose_port(self):

image = Image.windows().expose(8080)

async with Sandbox.ephemeral(image, local=True) as sb:

r = await sb.shell.run("echo exposed")

assert r.success

# ═══════════════════════════════════════════════════════════════════════════

# Android VM

# ═══════════════════════════════════════════════════════════════════════════

class TestAndroidApkInstall:

def setup_method(self):

skip_if_unsupported(Image.android())

async def test_apk_install(self):

"""Basic APK install — downloads F-Droid APK from URL."""

image = Image.android().apk_install("https://f-droid.org/F-Droid.apk")

async with Sandbox.ephemeral(image, local=True) as sb:

r = await sb.shell.run("pm list packages")

assert r.success

assert "org.fdroid" in r.stdout

async def test_apk_install_from_env(self):

import os

apk_path = os.environ.get("ANDROID_TEST_APK")

if not apk_path:

pytest.skip("ANDROID_TEST_APK not set")

image = Image.android().apk_install(apk_path)

async with Sandbox.ephemeral(image, local=True) as sb:

r = await sb.shell.run("pm list packages")

assert r.success

class TestAndroidPwaInstall:

def setup_method(self):

skip_if_unsupported(Image.android())

async def test_pwa_install(self):

"""PWA install from the android-example-gym-pwa-app manifest."""

import os

import tempfile

import urllib.request

manifest_url = os.environ.get(

"ANDROID_TEST_PWA_URL", "https://cuaai--todo-gym-web.modal.run/manifest.json"

)

keystore_url = "https://raw.githubusercontent.com/trycua/android-example-gym-pwa-app/main/android.keystore"

with tempfile.NamedTemporaryFile(suffix=".keystore", delete=False) as f:

urllib.request.urlretrieve(keystore_url, f.name)

keystore_path = f.name

image = Image.android().pwa_install(

manifest_url,

keystore=keystore_path,

keystore_alias="android",

keystore_password="android",

)

async with Sandbox.ephemeral(image, local=True) as sb:

r = await sb.shell.run("pm list packages")

assert r.success

class TestAndroidRun:

def setup_method(self):

skip_if_unsupported(Image.android())

async def test_run_adb_shell(self):

image = Image.android().run("echo android > /data/local/tmp/test.txt")

out = await _run(image, "cat /data/local/tmp/test.txt")

assert "android" in out

class TestAndroidEnv:

def setup_method(self):

skip_if_unsupported(Image.android())

async def test_env_variable(self):

image = Image.android().env(MY_VAR="android_test")

out = await _run(image, "echo $MY_VAR")

assert "android_test" in out