Catty Steve
2133f8b9b3
generation and pipeline scheduling - Add benchmark infrastructure (Criterion) for parser performance testing - Enhance decorator parsing with new types: Pipe, Parallel, Daemon, Health, and Unknown - Improve parser to track line numbers and handle nested braces correctly - Add implicit `@after` dependencies for pipeline functions in topological sort - Introduce `use_template` flag in builder to support trivial script execution - Update bake function signature to accept Cli parameters and privilege detection - Refactor error handling with structured `ScriptGenerationFailed` variant - Extend integration tests for privilege dropping, engine execution, and bake phase - Comment out daemon and apt modules temporarily for refactoring
1262 lines
45 KiB
Python
1262 lines
45 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Integration tests for workshop-baker prebake workflow.
|
|
|
|
These tests execute the complete prebake pipeline in a Docker container
|
|
running archlinux, using pre-built binaries from the host system.
|
|
|
|
Usage:
|
|
pytest tests/integration/test_prebake.py -v
|
|
pytest tests/integration/test_prebake.py -v -k "test_bootstrap" # Run specific test
|
|
"""
|
|
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import docker
|
|
import pytest
|
|
|
|
|
|
def docker_available():
|
|
"""Check if Docker daemon is available."""
|
|
try:
|
|
client = docker.from_env()
|
|
client.ping()
|
|
client.close()
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
class DockerContainer:
|
|
"""Context manager for Docker container lifecycle."""
|
|
|
|
def __init__(self, image: str, name: Optional[str] = None, **kwargs):
|
|
self.image = image
|
|
self.name = name or f"workshop-baker-test-{int(time.time())}"
|
|
self.container = None
|
|
self.client = docker.from_env()
|
|
self.host_dir = Path(tempfile.mkdtemp(prefix="workshop-baker-"))
|
|
self.pull_output = None
|
|
self.container_kwargs = kwargs
|
|
|
|
def __enter__(self):
|
|
try:
|
|
print(f"Checking for image {self.image}...")
|
|
self.client.images.get(self.image)
|
|
print(f"Image {self.image} found locally")
|
|
except docker.errors.ImageNotFound:
|
|
try:
|
|
print(f"Pulling image {self.image}...")
|
|
self.pull_output = self.client.images.pull(self.image)
|
|
print(f"Successfully pulled {self.image}")
|
|
except docker.errors.DockerException as e:
|
|
print(f"Docker exception: {e}")
|
|
raise
|
|
|
|
self.container = self.client.containers.run(
|
|
self.image,
|
|
detach=True,
|
|
name=self.name,
|
|
remove=True,
|
|
command="sleep infinity",
|
|
tty=True,
|
|
stdin_open=True,
|
|
security_opt=["seccomp=unconfined"],
|
|
cap_add=["SYS_ADMIN"],
|
|
**self.container_kwargs,
|
|
)
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
if self.container:
|
|
try:
|
|
self.container.stop(timeout=10)
|
|
except Exception as e:
|
|
print(f"Warning: Failed to stop container: {e}")
|
|
self.client.close()
|
|
return False
|
|
|
|
def exec_run(self, cmd: str, workdir: Optional[str] = None) -> tuple[int, str, str]:
|
|
"""Execute command in container, return (exit_code, stdout, stderr)."""
|
|
if self.container is None:
|
|
return -1, "", "Container not running"
|
|
result = self.container.exec_run(
|
|
cmd,
|
|
workdir=workdir,
|
|
demux=True,
|
|
)
|
|
exit_code = result.exit_code
|
|
stdout, stderr = result.output
|
|
return (
|
|
exit_code,
|
|
(stdout or b"").decode("utf-8", errors="replace"),
|
|
(stderr or b"").decode("utf-8", errors="replace"),
|
|
)
|
|
|
|
def put_archive(self, local_path: Path, container_path: str):
|
|
"""Copy local directory to container."""
|
|
if self.container is None:
|
|
raise RuntimeError("Container not running")
|
|
import tarfile
|
|
import io
|
|
|
|
tar_stream = io.BytesIO()
|
|
with tarfile.open(fileobj=tar_stream, mode="w") as tar:
|
|
tar.add(local_path, arcname=local_path.name)
|
|
tar_stream.seek(0)
|
|
|
|
self.container.put_archive(os.path.dirname(container_path), tar_stream)
|
|
|
|
def copy_file(self, host_path: Path, container_path: str):
|
|
"""Copy a single file to the container."""
|
|
if self.container is None:
|
|
raise RuntimeError("Container not running")
|
|
import tarfile
|
|
import io
|
|
|
|
tar_stream = io.BytesIO()
|
|
with tarfile.open(fileobj=tar_stream, mode="w") as tar:
|
|
tar.add(host_path, arcname=os.path.basename(container_path))
|
|
tar_stream.seek(0)
|
|
|
|
self.container.put_archive(os.path.dirname(container_path), tar_stream)
|
|
|
|
|
|
def check_user_exists(container: DockerContainer, username: str) -> bool:
|
|
"""Check if user <USER> exists in the container."""
|
|
exit_code, stdout, stderr = container.exec_run(f"id {username}")
|
|
return exit_code == 0
|
|
|
|
|
|
def check_user_home(container: DockerContainer, username: str) -> str | None:
|
|
"""Check user <USER>'s home directory. Returns path or None if not found."""
|
|
exit_code, stdout, stderr = container.exec_run(f"getent passwd {username}")
|
|
if exit_code != 0:
|
|
return None
|
|
parts = stdout.strip().split(":")
|
|
if len(parts) >= 6:
|
|
return parts[5]
|
|
return None
|
|
|
|
|
|
def check_directory_exists(container: DockerContainer, path: str) -> bool:
|
|
"""Check if directory <DIR> exists."""
|
|
exit_code, stdout, stderr = container.exec_run(f"test -d {path}")
|
|
return exit_code == 0
|
|
|
|
|
|
def check_file_exists(container: DockerContainer, path: str) -> bool:
|
|
"""Check if file <FILE> exists."""
|
|
exit_code, stdout, stderr = container.exec_run(f"test -f {path}")
|
|
return exit_code == 0
|
|
|
|
|
|
def check_file_content(
|
|
container: DockerContainer, path: str, pattern: str, fuzzy: bool = True
|
|
) -> bool:
|
|
"""Check if file <FILE> contains content matching pattern.
|
|
|
|
Args:
|
|
container: DockerContainer instance
|
|
path: file path in container
|
|
pattern: pattern to match
|
|
fuzzy: if True, use grep (partial match); if False, use exact match
|
|
"""
|
|
if not check_file_exists(container, path):
|
|
return False
|
|
|
|
if fuzzy:
|
|
exit_code, stdout, stderr = container.exec_run(f"grep -q '{pattern}' {path}")
|
|
return exit_code == 0
|
|
else:
|
|
exit_code, stdout, stderr = container.exec_run(f"grep -F '{pattern}' {path}")
|
|
return exit_code == 0
|
|
|
|
|
|
def check_sudo_permission(
|
|
container: DockerContainer, username: str, command: str
|
|
) -> tuple[bool, str]:
|
|
"""Check if user <USER> can/cannot sudo/doas to complete an operation.
|
|
|
|
Returns:
|
|
(can_execute, error_message)
|
|
"""
|
|
exit_code, stdout, stderr = container.exec_run(
|
|
f"sudo -u {username} -- sh -c '{command}'"
|
|
)
|
|
if exit_code == 0:
|
|
return True, ""
|
|
return False, stderr
|
|
|
|
|
|
def check_package_installed(container: DockerContainer, package: str) -> bool:
|
|
"""Check if software package <PACKAGE> is installed."""
|
|
exit_code, stdout, stderr = container.exec_run(
|
|
"sh -c 'command -v pacman >/dev/null 2>&1 && pacman -Q "
|
|
+ package
|
|
+ " >/dev/null 2>&1 || command -v apt >/dev/null 2>&1 && dpkg -l "
|
|
+ package
|
|
+ " >/dev/null 2>&1 || command -v rpm >/dev/null 2>&1 && rpm -q "
|
|
+ package
|
|
+ " >/dev/null 2>&1'"
|
|
)
|
|
return exit_code == 0
|
|
|
|
|
|
def check_env_var(
|
|
container: DockerContainer, var_name: str, expected_value: str = ""
|
|
) -> tuple[bool, str]:
|
|
"""Check environment variable <ENV> value.
|
|
|
|
Args:
|
|
container: DockerContainer instance
|
|
var_name: environment variable name
|
|
expected_value: if provided, check for exact match; if None, just check existence
|
|
|
|
Returns:
|
|
(matches, actual_value)
|
|
"""
|
|
exit_code, stdout, stderr = container.exec_run(f"echo ${var_name}")
|
|
if exit_code != 0:
|
|
return False, ""
|
|
|
|
actual_value = stdout.strip()
|
|
if expected_value is None:
|
|
return True, actual_value
|
|
|
|
return actual_value == expected_value, actual_value
|
|
|
|
|
|
class TestBootstrapUnit:
|
|
"""Unit tests for bootstrap stage - tests bootstrap.sh generation."""
|
|
|
|
IMAGE = "archlinux:latest"
|
|
PROJECT_ROOT = Path(__file__).parent.parent.parent
|
|
BINARY_NAME = "workshop-baker"
|
|
|
|
@pytest.fixture(scope="class")
|
|
def container_with_binary(self):
|
|
"""Provide a running archlinux container with pre-built binary copied."""
|
|
if not docker_available():
|
|
pytest.skip("Docker daemon is not available")
|
|
|
|
release_binary = self.PROJECT_ROOT / "target" / "release" / self.BINARY_NAME
|
|
debug_binary = self.PROJECT_ROOT / "target" / "debug" / self.BINARY_NAME
|
|
|
|
if release_binary.exists():
|
|
binary_path = release_binary
|
|
elif debug_binary.exists():
|
|
binary_path = debug_binary
|
|
else:
|
|
pytest.skip(
|
|
f"Binary {self.BINARY_NAME} not found. Run 'cargo build --release' first."
|
|
)
|
|
|
|
client = docker.from_env()
|
|
try:
|
|
client.images.get(self.IMAGE)
|
|
except docker.errors.ImageNotFound:
|
|
try:
|
|
print(f"Pulling image {self.IMAGE}...")
|
|
client.images.pull(self.IMAGE)
|
|
except docker.errors.DockerException as e:
|
|
pytest.skip(f"Could not pull image {self.IMAGE}: {e}")
|
|
except Exception as e:
|
|
pytest.skip(f"Could not access Docker: {e}")
|
|
finally:
|
|
client.close()
|
|
|
|
container_kwargs = {
|
|
"volumes": {
|
|
str(self.PROJECT_ROOT): {"bind": "/workshop-baker", "mode": "rw"},
|
|
},
|
|
}
|
|
|
|
container = DockerContainer(
|
|
self.IMAGE,
|
|
f"workshop-baker-unit-{int(time.time())}",
|
|
**container_kwargs,
|
|
)
|
|
|
|
with container:
|
|
container.copy_file(binary_path, f"/usr/local/bin/{self.BINARY_NAME}")
|
|
|
|
exit_code, stdout, stderr = container.exec_run(
|
|
f"chmod +x /usr/local/bin/{self.BINARY_NAME}"
|
|
)
|
|
if exit_code != 0:
|
|
pytest.fail(f"Failed to make binary executable: {stderr}")
|
|
|
|
bootstrap_sh = container.host_dir / "bootstrap.sh"
|
|
bootstrap_sh.write_text("#!/bin/bash\necho 'Bootstrap executed'\nexit 0\n")
|
|
container.copy_file(bootstrap_sh, "/bootstrap.sh")
|
|
|
|
exit_code, stdout, stderr = container.exec_run("chmod +x /bootstrap.sh")
|
|
if exit_code != 0:
|
|
pytest.fail(f"Failed to set bootstrap.sh executable: {stderr}")
|
|
|
|
yield container
|
|
|
|
def test_binary_help(self, container_with_binary: DockerContainer):
|
|
"""Verify binary runs and shows help."""
|
|
cmd = f"/usr/local/bin/{self.BINARY_NAME} --help"
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(cmd)
|
|
assert exit_code == 0, f"Binary help failed: {stderr}"
|
|
|
|
def test_prebake_subcommand_exists(self, container_with_binary: DockerContainer):
|
|
"""Verify prebake subcommand is recognized."""
|
|
cmd = f"/usr/local/bin/{self.BINARY_NAME} --help"
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(
|
|
cmd, workdir="/workshop-baker"
|
|
)
|
|
combined = stdout + stderr
|
|
assert "prebake" in combined.lower() or exit_code == 0
|
|
|
|
def test_bootstrap_script_runs(self, container_with_binary: DockerContainer):
|
|
"""Test that bootstrap.sh can be executed."""
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(
|
|
"/bootstrap.sh", workdir="/workshop-baker"
|
|
)
|
|
assert exit_code == 0, f"Bootstrap script failed: {stderr}"
|
|
assert "Bootstrap executed" in stdout
|
|
|
|
|
|
class TestPrebakeStages:
|
|
"""Integration tests for each prebake stage."""
|
|
|
|
IMAGE = "archlinux:latest"
|
|
PROJECT_ROOT = Path(__file__).parent.parent.parent
|
|
BINARY_NAME = "workshop-baker"
|
|
EXAMPLES_DIR = PROJECT_ROOT / "examples"
|
|
PREBAKE_YML = EXAMPLES_DIR / "prebake.yml"
|
|
|
|
@pytest.fixture(scope="class")
|
|
def container_with_binary(self):
|
|
"""Provide a running archlinux container with pre-built binary copied."""
|
|
if not docker_available():
|
|
pytest.skip("Docker daemon is not available")
|
|
|
|
release_binary = self.PROJECT_ROOT / "target" / "release" / self.BINARY_NAME
|
|
debug_binary = self.PROJECT_ROOT / "target" / "debug" / self.BINARY_NAME
|
|
|
|
if release_binary.exists():
|
|
binary_path = release_binary
|
|
elif debug_binary.exists():
|
|
binary_path = debug_binary
|
|
else:
|
|
pytest.skip(
|
|
f"Binary {self.BINARY_NAME} not found. Run 'cargo build --release' first."
|
|
)
|
|
|
|
client = docker.from_env()
|
|
try:
|
|
client.images.get(self.IMAGE)
|
|
except docker.errors.ImageNotFound:
|
|
try:
|
|
print(f"Pulling image {self.IMAGE}...")
|
|
client.images.pull(self.IMAGE)
|
|
except docker.errors.DockerException as e:
|
|
pytest.skip(f"Could not pull image {self.IMAGE}: {e}")
|
|
except Exception as e:
|
|
pytest.skip(f"Could not access Docker: {e}")
|
|
finally:
|
|
client.close()
|
|
|
|
container_kwargs = {
|
|
"volumes": {
|
|
str(self.PROJECT_ROOT): {"bind": "/workshop-baker", "mode": "rw"},
|
|
},
|
|
}
|
|
|
|
container = DockerContainer(
|
|
self.IMAGE,
|
|
f"workshop-baker-stages-{int(time.time())}",
|
|
**container_kwargs,
|
|
)
|
|
|
|
with container:
|
|
container.copy_file(binary_path, f"/usr/local/bin/{self.BINARY_NAME}")
|
|
|
|
exit_code, stdout, stderr = container.exec_run(
|
|
f"chmod +x /usr/local/bin/{self.BINARY_NAME}"
|
|
)
|
|
if exit_code != 0:
|
|
pytest.fail(f"Failed to make binary executable: {stderr}")
|
|
|
|
bootstrap_sh = container.host_dir / "bootstrap.sh"
|
|
bootstrap_sh.write_text(
|
|
"#!/bin/bash\n"
|
|
"set -e\n"
|
|
"echo 'Bootstrap stage: Creating user'\n"
|
|
"if command -v useradd &> /dev/null; then\n"
|
|
" useradd -m -s /bin/nologin vulcan 2>/dev/null || true\n"
|
|
"elif command -v adduser &> /dev/null; then\n"
|
|
" adduser -D vulcan 2>/dev/null || true\n"
|
|
"fi\n"
|
|
"echo 'Bootstrap stage: Setting up workspace'\n"
|
|
"mkdir -p /workspace\n"
|
|
"ln -sf /workspace /workspace_link 2>/dev/null || true\n"
|
|
"echo 'Bootstrap complete!'\n"
|
|
"exit 0\n"
|
|
)
|
|
container.copy_file(bootstrap_sh, "/bootstrap.sh")
|
|
|
|
exit_code, stdout, stderr = container.exec_run("chmod +x /bootstrap.sh")
|
|
if exit_code != 0:
|
|
pytest.fail(f"Failed to set bootstrap.sh executable: {stderr}")
|
|
|
|
yield container
|
|
|
|
def test_prebake_config_parsing(self, container_with_binary: DockerContainer):
|
|
"""Test that prebake.yml can be parsed without errors."""
|
|
cmd = (
|
|
f"/usr/local/bin/{self.BINARY_NAME} "
|
|
"--username test-user "
|
|
"--pipeline test-pipeline "
|
|
"--build-id test-build-001 "
|
|
"--dry-run "
|
|
"prebake examples/prebake.yml"
|
|
)
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(
|
|
cmd, workdir="/workshop-baker"
|
|
)
|
|
|
|
print(f"Config parsing:\nSTDOUT:\n{stdout}\nSTDERR:\n{stderr}")
|
|
|
|
combined = stdout + stderr
|
|
assert "error" not in combined.lower() or exit_code == 0, (
|
|
f"Config parsing failed: {stderr}"
|
|
)
|
|
|
|
def test_bootstrap_stage_executes(self, container_with_binary: DockerContainer):
|
|
"""Test bootstrap stage runs successfully."""
|
|
cmd = (
|
|
f"/usr/local/bin/{self.BINARY_NAME} "
|
|
"--username test-user "
|
|
"--pipeline test-pipeline "
|
|
"--build-id test-build-001 "
|
|
"prebake examples/prebake.yml"
|
|
)
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(
|
|
cmd, workdir="/workshop-baker"
|
|
)
|
|
|
|
print(f"Bootstrap stage:\nSTDOUT:\n{stdout}\nSTDERR:\n{stderr}")
|
|
|
|
assert exit_code == 0, (
|
|
f"Bootstrap stage failed with exit code {exit_code}: {stderr}"
|
|
)
|
|
|
|
def test_early_hook_executes(self, container_with_binary: DockerContainer):
|
|
"""Test early hook runs successfully."""
|
|
cmd = (
|
|
f"/usr/local/bin/{self.BINARY_NAME} "
|
|
"--username test-user "
|
|
"--pipeline test-pipeline "
|
|
"--build-id test-build-001 "
|
|
"prebake examples/prebake.yml"
|
|
)
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(
|
|
cmd, workdir="/workshop-baker"
|
|
)
|
|
|
|
print(f"Early hook:\nSTDOUT:\n{stdout}\nSTDERR:\n{stderr}")
|
|
|
|
assert exit_code == 0, f"Early hook failed with exit code {exit_code}: {stderr}"
|
|
|
|
def test_late_hook_executes(self, container_with_binary: DockerContainer):
|
|
"""Test late hook runs successfully."""
|
|
cmd = (
|
|
f"/usr/local/bin/{self.BINARY_NAME} "
|
|
"--username test-user "
|
|
"--pipeline test-pipeline "
|
|
"--build-id test-build-001 "
|
|
"prebake examples/prebake.yml"
|
|
)
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(
|
|
cmd, workdir="/workshop-baker"
|
|
)
|
|
|
|
print(f"Late hook:\nSTDOUT:\n{stdout}\nSTDERR:\n{stderr}")
|
|
|
|
assert exit_code == 0, f"Late hook failed with exit code {exit_code}: {stderr}"
|
|
|
|
def test_full_prebake_pipeline(self, container_with_binary: DockerContainer):
|
|
"""Test complete prebake pipeline from start to finish."""
|
|
cmd = (
|
|
f"/usr/local/bin/{self.BINARY_NAME} "
|
|
"--username test-user "
|
|
"--pipeline test-pipeline "
|
|
"--build-id test-build-001 "
|
|
"prebake examples/prebake.yml"
|
|
)
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(
|
|
cmd, workdir="/workshop-baker"
|
|
)
|
|
|
|
print(f"Full pipeline:\nSTDOUT:\n{stdout}\nSTDERR:\n{stderr}")
|
|
|
|
assert exit_code == 0, (
|
|
f"Full prebake pipeline failed with exit code {exit_code}: {stderr}"
|
|
)
|
|
|
|
def test_bootstrap_creates_user(self, container_with_binary: DockerContainer):
|
|
assert check_user_exists(container_with_binary, "vulcan")
|
|
|
|
def test_bootstrap_creates_workspace(self, container_with_binary: DockerContainer):
|
|
assert check_directory_exists(container_with_binary, "/workspace")
|
|
|
|
|
|
class TestPrebakeIntegration:
|
|
"""End-to-end integration tests for prebake workflow."""
|
|
|
|
IMAGE = "archlinux:latest"
|
|
PROJECT_ROOT = Path(__file__).parent.parent.parent
|
|
BINARY_NAME = "workshop-baker"
|
|
|
|
@pytest.fixture(scope="class")
|
|
def container_with_binary(self):
|
|
"""Provide a running archlinux container with pre-built binary copied."""
|
|
if not docker_available():
|
|
pytest.skip("Docker daemon is not available")
|
|
|
|
release_binary = self.PROJECT_ROOT / "target" / "release" / self.BINARY_NAME
|
|
debug_binary = self.PROJECT_ROOT / "target" / "debug" / self.BINARY_NAME
|
|
|
|
if release_binary.exists():
|
|
binary_path = release_binary
|
|
elif debug_binary.exists():
|
|
binary_path = debug_binary
|
|
else:
|
|
pytest.skip(
|
|
f"Binary {self.BINARY_NAME} not found. Run 'cargo build --release' first."
|
|
)
|
|
|
|
client = docker.from_env()
|
|
try:
|
|
client.images.get(self.IMAGE)
|
|
except docker.errors.ImageNotFound:
|
|
try:
|
|
print(f"Pulling image {self.IMAGE}...")
|
|
client.images.pull(self.IMAGE)
|
|
except docker.errors.DockerException as e:
|
|
pytest.skip(f"Could not pull image {self.IMAGE}: {e}")
|
|
except Exception as e:
|
|
pytest.skip(f"Could not access Docker: {e}")
|
|
finally:
|
|
client.close()
|
|
|
|
container_kwargs = {
|
|
"volumes": {
|
|
str(self.PROJECT_ROOT): {"bind": "/workshop-baker", "mode": "rw"},
|
|
},
|
|
}
|
|
|
|
container = DockerContainer(
|
|
self.IMAGE,
|
|
None,
|
|
**container_kwargs,
|
|
)
|
|
|
|
with container:
|
|
container.copy_file(binary_path, f"/usr/local/bin/{self.BINARY_NAME}")
|
|
|
|
exit_code, stdout, stderr = container.exec_run(
|
|
f"chmod +x /usr/local/bin/{self.BINARY_NAME}"
|
|
)
|
|
if exit_code != 0:
|
|
pytest.fail(f"Failed to make binary executable: {stderr}")
|
|
|
|
bootstrap_sh = container.host_dir / "bootstrap.sh"
|
|
bootstrap_sh.write_text(
|
|
"#!/bin/bash\nset -e\necho 'Bootstrap executed successfully'\nexit 0\n"
|
|
)
|
|
container.copy_file(bootstrap_sh, "/bootstrap.sh")
|
|
|
|
exit_code, stdout, stderr = container.exec_run("chmod +x /bootstrap.sh")
|
|
if exit_code != 0:
|
|
pytest.fail(f"Failed to set bootstrap.sh executable: {stderr}")
|
|
|
|
yield container
|
|
|
|
def test_container_network_connectivity(
|
|
self, container_with_binary: DockerContainer
|
|
):
|
|
"""Test that container has network connectivity."""
|
|
exit_code, stdout, stderr = container_with_binary.exec_run("ping -c 1 8.8.8.8")
|
|
assert exit_code == 0, f"Container network connectivity failed: {stderr}"
|
|
|
|
def test_binary_executable_in_container(
|
|
self, container_with_binary: DockerContainer
|
|
):
|
|
"""Test that the pre-built binary is executable in the container."""
|
|
cmd = f"/usr/local/bin/{self.BINARY_NAME} --version"
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(cmd)
|
|
print(f"Binary version:\nSTDOUT:\n{stdout}\nSTDERR:\n{stderr}")
|
|
assert exit_code == 0, f"Binary execution failed: {stderr}"
|
|
|
|
def test_prebake_command_in_container(self, container_with_binary: DockerContainer):
|
|
"""Test that prebake command works in the container."""
|
|
cmd = (
|
|
f"/usr/local/bin/{self.BINARY_NAME} "
|
|
"--username test-user "
|
|
"--pipeline test-pipeline "
|
|
"--build-id test-build-001 "
|
|
"prebake examples/prebake.yml"
|
|
)
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(
|
|
cmd, workdir="/workshop-baker"
|
|
)
|
|
|
|
print(f"Prebake command:\nSTDOUT:\n{stdout}\nSTDERR:\n{stderr}")
|
|
|
|
combined = stdout + stderr
|
|
assert exit_code == 0 or "bootstrap" in combined.lower(), (
|
|
f"Prebake command failed with exit code {exit_code}: {stderr}"
|
|
)
|
|
|
|
|
|
class TestPrebakeConfigValidation:
|
|
"""Tests for prebake.yml configuration validation."""
|
|
|
|
PROJECT_ROOT = Path(__file__).parent.parent.parent
|
|
EXAMPLES_DIR = PROJECT_ROOT / "examples"
|
|
PREBAKE_YML = EXAMPLES_DIR / "prebake.yml"
|
|
|
|
def test_prebake_yml_exists(self):
|
|
"""Verify prebake.yml exists and is readable."""
|
|
assert self.PREBAKE_YML.exists(), f"prebake.yml not found at {self.PREBAKE_YML}"
|
|
assert self.PREBAKE_YML.is_file(), f"prebake.yml is not a file"
|
|
|
|
def test_prebake_yml_valid_yaml(self):
|
|
"""Verify prebake.yml is valid YAML."""
|
|
import yaml
|
|
|
|
content = self.PREBAKE_YML.read_text()
|
|
try:
|
|
data = yaml.safe_load(content)
|
|
assert data is not None, "prebake.yml is empty"
|
|
assert isinstance(data, dict), "prebake.yml root should be a dictionary"
|
|
except yaml.YAMLError as e:
|
|
pytest.fail(f"Invalid YAML in prebake.yml: {e}")
|
|
|
|
def test_prebake_yml_required_fields(self):
|
|
"""Verify prebake.yml has all required top-level fields."""
|
|
import yaml
|
|
|
|
content = self.PREBAKE_YML.read_text()
|
|
data = yaml.safe_load(content)
|
|
|
|
required_fields = ["version", "environment", "dependencies"]
|
|
for field in required_fields:
|
|
assert field in data, f"Missing required field: {field}"
|
|
|
|
def test_prebake_yml_version_format(self):
|
|
"""Verify version field is properly formatted."""
|
|
import yaml
|
|
|
|
content = self.PREBAKE_YML.read_text()
|
|
data = yaml.safe_load(content)
|
|
|
|
version = data.get("version", "")
|
|
assert re.match(r"^\d+\.\d+(\.\d+)?$", str(version)), (
|
|
f"Invalid version format: {version}"
|
|
)
|
|
|
|
def test_prebake_yml_environment_builder(self):
|
|
"""Verify environment.builder is set."""
|
|
import yaml
|
|
|
|
content = self.PREBAKE_YML.read_text()
|
|
data = yaml.safe_load(content)
|
|
|
|
builder = data.get("environment", {}).get("builder", "")
|
|
assert builder in ["docker", "firecracker", "custom", "baremetal"], (
|
|
f"Invalid builder type: {builder}"
|
|
)
|
|
|
|
def test_prebake_yml_dependencies_structure(self):
|
|
"""Verify dependencies section has expected structure."""
|
|
import yaml
|
|
|
|
content = self.PREBAKE_YML.read_text()
|
|
data = yaml.safe_load(content)
|
|
|
|
deps = data.get("dependencies", {})
|
|
if deps:
|
|
has_system = "system" in deps
|
|
has_user = "user" in deps
|
|
assert has_system or has_user, (
|
|
"Dependencies must have at least one of: system, user"
|
|
)
|
|
|
|
def test_prebake_yml_hooks_structure(self):
|
|
"""Verify hooks section has valid structure if present."""
|
|
import yaml
|
|
|
|
content = self.PREBAKE_YML.read_text()
|
|
data = yaml.safe_load(content)
|
|
|
|
hooks = data.get("hooks", {})
|
|
if hooks:
|
|
valid_hook_types = ["early", "late"]
|
|
for hook_type in hooks.keys():
|
|
assert hook_type in valid_hook_types, f"Invalid hook type: {hook_type}"
|
|
|
|
for hook_type, hook_list in hooks.items():
|
|
assert isinstance(hook_list, list), f"Hook {hook_type} should be a list"
|
|
|
|
|
|
class TestPrebakeExamples:
|
|
"""Test suite for example files in the project."""
|
|
|
|
PROJECT_ROOT = Path(__file__).parent.parent.parent
|
|
EXAMPLES_DIR = PROJECT_ROOT / "examples"
|
|
|
|
def test_examples_directory_exists(self):
|
|
"""Verify examples directory exists."""
|
|
assert self.EXAMPLES_DIR.exists(), f"Examples directory not found"
|
|
assert self.EXAMPLES_DIR.is_dir(), f"Examples path is not a directory"
|
|
|
|
def test_dockerfile_example_exists(self):
|
|
"""Verify Dockerfile example exists."""
|
|
dockerfile = self.EXAMPLES_DIR / "Dockerfile_1"
|
|
assert dockerfile.exists(), f"Dockerfile example not found at {dockerfile}"
|
|
|
|
def test_prebake_build_yml_exists(self):
|
|
"""Verify prebake_build.yml example exists."""
|
|
prebake_build = self.EXAMPLES_DIR / "prebake_build.yml"
|
|
if prebake_build.exists():
|
|
import yaml
|
|
|
|
content = prebake_build.read_text()
|
|
try:
|
|
yaml.safe_load(content)
|
|
except yaml.YAMLError as e:
|
|
pytest.fail(f"Invalid YAML in prebake_build.yml: {e}")
|
|
|
|
|
|
class TestEngineInDocker:
|
|
"""Engine integration tests running in Docker container.
|
|
|
|
These tests verify the engine can execute commands correctly inside a container.
|
|
If Docker is not available, tests are skipped with appropriate logging.
|
|
"""
|
|
|
|
IMAGE = "archlinux:latest"
|
|
PROJECT_ROOT = Path(__file__).parent.parent.parent
|
|
BINARY_NAME = "workshop-baker"
|
|
|
|
@pytest.fixture(scope="class")
|
|
def container_with_binary(self):
|
|
if not docker_available():
|
|
pytest.skip("Docker未启动")
|
|
|
|
release_binary = self.PROJECT_ROOT / "target" / "release" / self.BINARY_NAME
|
|
debug_binary = self.PROJECT_ROOT / "target" / "debug" / self.BINARY_NAME
|
|
|
|
if release_binary.exists():
|
|
binary_path = release_binary
|
|
elif debug_binary.exists():
|
|
binary_path = debug_binary
|
|
else:
|
|
pytest.skip(
|
|
f"Binary {self.BINARY_NAME} not found. Run 'cargo build --release' first."
|
|
)
|
|
|
|
client = docker.from_env()
|
|
try:
|
|
client.images.get(self.IMAGE)
|
|
except docker.errors.ImageNotFound:
|
|
try:
|
|
print(f"Pulling image {self.IMAGE}...")
|
|
client.images.pull(self.IMAGE)
|
|
except docker.errors.DockerException as e:
|
|
pytest.skip(f"Could not pull image {self.IMAGE}: {e}")
|
|
except Exception as e:
|
|
pytest.skip(f"Could not access Docker: {e}")
|
|
finally:
|
|
client.close()
|
|
|
|
container_kwargs = {
|
|
"volumes": {
|
|
str(self.PROJECT_ROOT): {"bind": "/workshop-baker", "mode": "rw"},
|
|
},
|
|
}
|
|
|
|
container = DockerContainer(
|
|
self.IMAGE,
|
|
f"workshop-baker-engine-{int(time.time())}",
|
|
**container_kwargs,
|
|
)
|
|
|
|
with container:
|
|
container.copy_file(binary_path, f"/usr/local/bin/{self.BINARY_NAME}")
|
|
|
|
exit_code, stdout, stderr = container.exec_run(
|
|
f"chmod +x /usr/local/bin/{self.BINARY_NAME}"
|
|
)
|
|
if exit_code != 0:
|
|
pytest.fail(f"Failed to make binary executable: {stderr}")
|
|
|
|
yield container
|
|
|
|
def test_execute_echo_outputs_correct_message(
|
|
self, container_with_binary: DockerContainer
|
|
):
|
|
cmd = "echo 'hello world'"
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(cmd)
|
|
assert exit_code == 0, f"echo failed: {stderr}"
|
|
assert "hello world" in stdout
|
|
|
|
def test_execute_true_returns_zero_exit_code(
|
|
self, container_with_binary: DockerContainer
|
|
):
|
|
cmd = "true"
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(cmd)
|
|
assert exit_code == 0, f"true command failed"
|
|
|
|
def test_execute_false_returns_nonzero_exit_code(
|
|
self, container_with_binary: DockerContainer
|
|
):
|
|
cmd = "false"
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(cmd)
|
|
assert exit_code != 0, f"false should return non-zero exit code"
|
|
|
|
def test_execute_ls_lists_directory(self, container_with_binary: DockerContainer):
|
|
cmd = "ls -la /tmp"
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(cmd)
|
|
assert exit_code == 0, f"ls failed: {stderr}"
|
|
|
|
def test_execute_date_command(self, container_with_binary: DockerContainer):
|
|
cmd = "date"
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(cmd)
|
|
assert exit_code == 0, f"date failed: {stderr}"
|
|
assert len(stdout.strip()) > 0
|
|
|
|
def test_execute_invalid_command_returns_error(
|
|
self, container_with_binary: DockerContainer
|
|
):
|
|
cmd = "this_command_definitely_does_not_exist_12345"
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(cmd)
|
|
assert exit_code != 0, f"invalid command should fail"
|
|
|
|
def test_execute_custom_env_var(self, container_with_binary: DockerContainer):
|
|
exit_code, stdout, stderr = container_with_binary.exec_run("env")
|
|
assert exit_code == 0, f"env failed: {stderr}"
|
|
|
|
def test_execute_working_directory(self, container_with_binary: DockerContainer):
|
|
cmd = "pwd"
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(cmd, workdir="/tmp")
|
|
assert exit_code == 0, f"pwd failed: {stderr}"
|
|
|
|
def test_execute_captures_stderr(self, container_with_binary: DockerContainer):
|
|
cmd = "bash -c 'echo error >&2'"
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(cmd)
|
|
assert exit_code == 0, f"stderr capture failed"
|
|
assert "error" in stderr
|
|
|
|
|
|
class TestBake:
|
|
"""Bake integration tests.
|
|
|
|
Tests the bake phase with trivial=true, use_template=true.
|
|
Uses bake_base.sh from workshop-pipeline/examples/.workshop/
|
|
"""
|
|
|
|
IMAGE = "archlinux:latest"
|
|
PROJECT_ROOT = Path(__file__).parent.parent.parent
|
|
BINARY_NAME = "workshop-baker"
|
|
BAKE_BASE = (
|
|
PROJECT_ROOT.parent
|
|
/ "workshop-pipeline"
|
|
/ "examples"
|
|
/ ".workshop"
|
|
/ "bake_base.sh"
|
|
)
|
|
|
|
@pytest.fixture(scope="class")
|
|
def container_with_binary(self):
|
|
if not docker_available():
|
|
pytest.skip("Docker未启动")
|
|
|
|
release_binary = self.PROJECT_ROOT / "target" / "release" / self.BINARY_NAME
|
|
debug_binary = self.PROJECT_ROOT / "target" / "debug" / self.BINARY_NAME
|
|
|
|
if release_binary.exists():
|
|
binary_path = release_binary
|
|
elif debug_binary.exists():
|
|
binary_path = debug_binary
|
|
else:
|
|
pytest.skip(
|
|
f"Binary {self.BINARY_NAME} not found. Run 'cargo build --release' first."
|
|
)
|
|
|
|
client = docker.from_env()
|
|
try:
|
|
client.images.get(self.IMAGE)
|
|
except docker.errors.ImageNotFound:
|
|
try:
|
|
print(f"Pulling image {self.IMAGE}...")
|
|
client.images.pull(self.IMAGE)
|
|
except docker.errors.DockerException as e:
|
|
pytest.skip(f"Could not pull image {self.IMAGE}: {e}")
|
|
except Exception as e:
|
|
pytest.skip(f"Could not access Docker: {e}")
|
|
finally:
|
|
client.close()
|
|
|
|
container_kwargs = {
|
|
"volumes": {
|
|
str(self.PROJECT_ROOT): {"bind": "/workshop-baker", "mode": "rw"},
|
|
},
|
|
}
|
|
|
|
container = DockerContainer(
|
|
self.IMAGE,
|
|
f"workshop-baker-bake-{int(time.time())}",
|
|
**container_kwargs,
|
|
)
|
|
|
|
with container:
|
|
container.copy_file(binary_path, f"/usr/local/bin/{self.BINARY_NAME}")
|
|
|
|
exit_code, stdout, stderr = container.exec_run(
|
|
f"chmod +x /usr/local/bin/{self.BINARY_NAME}"
|
|
)
|
|
if exit_code != 0:
|
|
pytest.fail(f"Failed to make binary executable: {stderr}")
|
|
|
|
if not self.BAKE_BASE.exists():
|
|
pytest.skip(f"bake_base.sh not found at {self.BAKE_BASE}")
|
|
container.copy_file(self.BAKE_BASE, "/bake_base.sh")
|
|
|
|
exit_code, stdout, stderr = container.exec_run("chmod +x /bake_base.sh")
|
|
if exit_code != 0:
|
|
pytest.fail(f"Failed to make bake_base.sh executable: {stderr}")
|
|
|
|
exit_code, stdout, stderr = container.exec_run("mkdir -p /workspace")
|
|
if exit_code != 0:
|
|
pytest.fail(f"Failed to create /workspace: {stderr}")
|
|
|
|
yield container
|
|
|
|
def test_bake_base_exists(self, container_with_binary: DockerContainer):
|
|
assert check_file_exists(container_with_binary, "/bake_base.sh")
|
|
|
|
def test_bake_binary_exists(self, container_with_binary: DockerContainer):
|
|
cmd = f"/usr/local/bin/{self.BINARY_NAME} --help"
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(cmd)
|
|
assert exit_code == 0 or "bake" in (stdout + stderr).lower()
|
|
|
|
def test_bake_simple_echo(self, container_with_binary: DockerContainer):
|
|
script_sh = container_with_binary.host_dir / "script.sh"
|
|
script_sh.write_text("echo 'Hello from bake'\n")
|
|
container_with_binary.copy_file(script_sh, "/script.sh")
|
|
|
|
cmd = (
|
|
f"/usr/local/bin/{self.BINARY_NAME} "
|
|
"--username test-user "
|
|
"--pipeline test-pipeline "
|
|
"--build-id test-build-001 "
|
|
"--dry-run "
|
|
f"bake /script.sh --bake-base /bake_base.sh --prebake /workshop-baker/examples/prebake.yml"
|
|
)
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(
|
|
cmd, workdir="/workshop-baker"
|
|
)
|
|
print(f"Bake simple echo:\nSTDOUT:\n{stdout}\nSTDERR:\n{stderr}")
|
|
assert "error" not in (stdout + stderr).lower() or exit_code == 0
|
|
|
|
def test_bake_executes_script_content(self, container_with_binary: DockerContainer):
|
|
script_sh = container_with_binary.host_dir / "script.sh"
|
|
script_sh.write_text("echo 'Bake test output'\n")
|
|
container_with_binary.copy_file(script_sh, "/script.sh")
|
|
|
|
cmd = (
|
|
f"/usr/local/bin/{self.BINARY_NAME} "
|
|
"--username test-user "
|
|
"--pipeline test-pipeline "
|
|
"--build-id test-build-001 "
|
|
f"bake /script.sh --bake-base /bake_base.sh --prebake /workshop-baker/examples/prebake.yml"
|
|
)
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(
|
|
cmd, workdir="/workshop-baker"
|
|
)
|
|
print(f"Bake executes:\nSTDOUT:\n{stdout}\nSTDERR:\n{stderr}")
|
|
assert exit_code == 0, f"Bake failed: {stderr}"
|
|
|
|
|
|
class TestPrivilegeDropping:
|
|
"""Tests for drop-after privilege dropping mechanism.
|
|
|
|
Verifies that privileges are correctly dropped at the specified stage boundary.
|
|
The drop-after setting determines when the process transitions from root to non-root.
|
|
|
|
Stage order: Init < Bootstrap < EarlyHook < DepsSystem < DepsUser < LateHook < Ready
|
|
If drop-after: LateHook, privileges are dropped AFTER LateHook completes.
|
|
If drop-after: DepsSystem (default), privileges are dropped AFTER DepsSystem completes.
|
|
"""
|
|
|
|
IMAGE = "archlinux:latest"
|
|
PROJECT_ROOT = Path(__file__).parent.parent.parent
|
|
BINARY_NAME = "workshop-baker"
|
|
|
|
@pytest.fixture(scope="class")
|
|
def container_with_binary(self):
|
|
if not docker_available():
|
|
pytest.skip("Docker未启动")
|
|
|
|
release_binary = self.PROJECT_ROOT / "target" / "release" / self.BINARY_NAME
|
|
debug_binary = self.PROJECT_ROOT / "target" / "debug" / self.BINARY_NAME
|
|
|
|
if release_binary.exists():
|
|
binary_path = release_binary
|
|
elif debug_binary.exists():
|
|
binary_path = debug_binary
|
|
else:
|
|
pytest.skip(
|
|
f"Binary {self.BINARY_NAME} not found. Run 'cargo build --release' first."
|
|
)
|
|
|
|
client = docker.from_env()
|
|
try:
|
|
client.images.get(self.IMAGE)
|
|
except docker.errors.ImageNotFound:
|
|
try:
|
|
print(f"Pulling image {self.IMAGE}...")
|
|
client.images.pull(self.IMAGE)
|
|
except docker.errors.DockerException as e:
|
|
pytest.skip(f"Could not pull image {self.IMAGE}: {e}")
|
|
except Exception as e:
|
|
pytest.skip(f"Could not access Docker: {e}")
|
|
finally:
|
|
client.close()
|
|
|
|
container_kwargs = {
|
|
"volumes": {
|
|
str(self.PROJECT_ROOT): {"bind": "/workshop-baker", "mode": "rw"},
|
|
},
|
|
"user": "0:0",
|
|
}
|
|
|
|
container = DockerContainer(
|
|
self.IMAGE,
|
|
f"workshop-baker-dropafter-{int(time.time())}",
|
|
**container_kwargs,
|
|
)
|
|
|
|
with container:
|
|
container.copy_file(binary_path, f"/usr/local/bin/{self.BINARY_NAME}")
|
|
|
|
exit_code, stdout, stderr = container.exec_run(
|
|
f"chmod +x /usr/local/bin/{self.BINARY_NAME}"
|
|
)
|
|
if exit_code != 0:
|
|
pytest.fail(f"Failed to make binary executable: {stderr}")
|
|
|
|
bootstrap_sh = container.host_dir / "bootstrap.sh"
|
|
bootstrap_sh.write_text(
|
|
"#!/bin/bash\n"
|
|
"set -e\n"
|
|
"if ! id -u testuser >/dev/null 2>&1; then\n"
|
|
" useradd -m -s /bin/bash testuser\n"
|
|
"fi\n"
|
|
"if ! id -u vulcan >/dev/null 2>&1; then\n"
|
|
" useradd -m -s /bin/nologin vulcan\n"
|
|
"fi\n"
|
|
"mkdir -p /workspace\n"
|
|
"chmod 777 /workspace\n"
|
|
"exit 0\n"
|
|
)
|
|
container.copy_file(bootstrap_sh, "/bootstrap.sh")
|
|
|
|
exit_code, stdout, stderr = container.exec_run("chmod +x /bootstrap.sh")
|
|
if exit_code != 0:
|
|
pytest.fail(f"Failed to set bootstrap.sh executable: {stderr}")
|
|
|
|
yield container
|
|
|
|
def _create_prebake_yml(
|
|
self, container: DockerContainer, drop_after: str, hook_cmd: str, hook_name: str
|
|
) -> Path:
|
|
prebake_content = f"""version: "1.0"
|
|
environment:
|
|
builder: "docker"
|
|
docker:
|
|
image: "archlinux:latest"
|
|
security:
|
|
drop-after: "{drop_after}"
|
|
hooks:
|
|
{hook_name}:
|
|
- command: "{hook_cmd}"
|
|
"""
|
|
prebake_path = container.host_dir / "test_prebake.yml"
|
|
prebake_path.write_text(prebake_content)
|
|
return prebake_path
|
|
|
|
def test_drop_after_late_hook_early_runs_privileged(
|
|
self, container_with_binary: DockerContainer
|
|
):
|
|
"""With drop-after: LateHook, EarlyHook runs as root (UID 0)."""
|
|
prebake_path = self._create_prebake_yml(
|
|
container_with_binary,
|
|
drop_after="LateHook",
|
|
hook_cmd="echo $(id -u) > /workspace/hook_uid.txt",
|
|
hook_name="early",
|
|
)
|
|
container_with_binary.copy_file(prebake_path, "/test_prebake.yml")
|
|
|
|
cmd = (
|
|
f"/usr/local/bin/{self.BINARY_NAME} "
|
|
"--username testuser "
|
|
"--pipeline test-pipeline "
|
|
"--build-id test-build-001 "
|
|
"prebake /test_prebake.yml"
|
|
)
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(
|
|
cmd, workdir="/workshop-baker"
|
|
)
|
|
|
|
print(f"drop-after: LateHook, early hook exit: {exit_code}")
|
|
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(
|
|
"cat /workspace/hook_uid.txt"
|
|
)
|
|
uid_str = stdout.strip() if stdout else ""
|
|
print(f"Early hook UID from file: {uid_str}")
|
|
assert uid_str == "0", (
|
|
f"EarlyHook should run as root (UID 0), but got {uid_str}"
|
|
)
|
|
|
|
def test_drop_after_late_hook_late_runs_unprivileged(
|
|
self, container_with_binary: DockerContainer
|
|
):
|
|
"""With drop-after: LateHook, LateHook runs as non-root (non-UID 0)."""
|
|
container_with_binary.exec_run("rm -f /workspace/hook_uid.txt")
|
|
|
|
prebake_path = self._create_prebake_yml(
|
|
container_with_binary,
|
|
drop_after="LateHook",
|
|
hook_cmd="echo $(id -u) > /workspace/hook_uid.txt",
|
|
hook_name="late",
|
|
)
|
|
container_with_binary.copy_file(prebake_path, "/test_prebake.yml")
|
|
|
|
cmd = (
|
|
f"/usr/local/bin/{self.BINARY_NAME} "
|
|
"--username testuser "
|
|
"--pipeline test-pipeline "
|
|
"--build-id test-build-001 "
|
|
"prebake /test_prebake.yml"
|
|
)
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(
|
|
cmd, workdir="/workshop-baker"
|
|
)
|
|
|
|
print(f"drop-after: LateHook, late hook exit: {exit_code}")
|
|
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(
|
|
"cat /workspace/hook_uid.txt"
|
|
)
|
|
uid_str = stdout.strip() if stdout else ""
|
|
print(f"Late hook UID from file: {uid_str}")
|
|
uid = int(uid_str)
|
|
assert uid != 0, f"LateHook should run as non-root, but got UID {uid}"
|
|
|
|
def test_drop_after_depssystem_early_runs_privileged(
|
|
self, container_with_binary: DockerContainer
|
|
):
|
|
"""With drop-after: DepsSystem (default), EarlyHook runs as root (UID 0)."""
|
|
container_with_binary.exec_run("rm -f /workspace/hook_uid.txt")
|
|
|
|
prebake_path = self._create_prebake_yml(
|
|
container_with_binary,
|
|
drop_after="DepsSystem",
|
|
hook_cmd="echo $(id -u) > /workspace/hook_uid.txt",
|
|
hook_name="early",
|
|
)
|
|
container_with_binary.copy_file(prebake_path, "/test_prebake.yml")
|
|
|
|
cmd = (
|
|
f"/usr/local/bin/{self.BINARY_NAME} "
|
|
"--username testuser "
|
|
"--pipeline test-pipeline "
|
|
"--build-id test-build-001 "
|
|
"prebake /test_prebake.yml"
|
|
)
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(
|
|
cmd, workdir="/workshop-baker"
|
|
)
|
|
|
|
print(f"drop-after: DepsSystem, early hook exit: {exit_code}")
|
|
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(
|
|
"cat /workspace/hook_uid.txt"
|
|
)
|
|
uid_str = stdout.strip() if stdout else ""
|
|
print(f"Early hook UID from file: {uid_str}")
|
|
assert uid_str == "0", (
|
|
f"EarlyHook should run as root (UID 0), but got {uid_str}"
|
|
)
|
|
|
|
def test_drop_after_depssystem_late_runs_unprivileged(
|
|
self, container_with_binary: DockerContainer
|
|
):
|
|
"""With drop-after: DepsSystem (default), LateHook runs as non-root (non-UID 0)."""
|
|
container_with_binary.exec_run("rm -f /workspace/hook_uid.txt")
|
|
|
|
prebake_path = self._create_prebake_yml(
|
|
container_with_binary,
|
|
drop_after="DepsSystem",
|
|
hook_cmd="echo $(id -u) > /workspace/hook_uid.txt",
|
|
hook_name="late",
|
|
)
|
|
container_with_binary.copy_file(prebake_path, "/test_prebake.yml")
|
|
|
|
cmd = (
|
|
f"/usr/local/bin/{self.BINARY_NAME} "
|
|
"--username testuser "
|
|
"--pipeline test-pipeline "
|
|
"--build-id test-build-001 "
|
|
"prebake /test_prebake.yml"
|
|
)
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(
|
|
cmd, workdir="/workshop-baker"
|
|
)
|
|
|
|
print(f"drop-after: DepsSystem, late hook exit: {exit_code}")
|
|
|
|
exit_code, stdout, stderr = container_with_binary.exec_run(
|
|
"cat /workspace/hook_uid.txt"
|
|
)
|
|
uid_str = stdout.strip() if stdout else ""
|
|
print(f"Late hook UID from file: {uid_str}")
|
|
uid = int(uid_str)
|
|
assert uid != 0, f"LateHook should run as non-root, but got UID {uid}"
|
|
|
|
|
|
def run_integration_tests():
|
|
"""Run integration tests with Docker."""
|
|
import sys
|
|
|
|
pytest.main(
|
|
[
|
|
__file__,
|
|
"-v",
|
|
"--tb=short",
|
|
"-p",
|
|
"no:warnings",
|
|
]
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run_integration_tests()
|