Catty Steve
262 lines
11 KiB
Python
262 lines
11 KiB
Python
import os
|
|
import json
|
|
import pytest
|
|
|
|
from probes import user, file, package
|
|
|
|
|
|
FIXTURES = os.path.join(os.path.dirname(__file__), "fixtures", "prebake")
|
|
|
|
|
|
@pytest.mark.prebake
|
|
class TestPrebakeConfigValidation:
|
|
|
|
def test_invalid_version_fails(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "invalid_version.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=60)
|
|
assert not result["success"], f"Expected prebake to fail with invalid version"
|
|
assert "version" in result["stderr"].lower() or "Version" in result["stderr"]
|
|
|
|
def test_missing_required_field_fails(self, run_baker, work_dir):
|
|
missing_env = os.path.join(work_dir, "missing_env.yml")
|
|
with open(missing_env, "w") as f:
|
|
f.write('version: "1.0"\n')
|
|
result = run_baker("prebake", missing_env, cwd=work_dir, timeout=60)
|
|
assert not result["success"]
|
|
|
|
def test_minimal_config_succeeds(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "minimal.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=60)
|
|
assert result["success"], f"Minimal prebake failed: {result['stderr']}"
|
|
|
|
|
|
@pytest.mark.prebake
|
|
class TestPrebakeBootstrap:
|
|
|
|
def test_bootstrap_creates_vulcan_user(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "bootstrap_only.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=60)
|
|
assert result["success"], f"prebake failed: {result['stderr']}"
|
|
assert user.user_exists("vulcan")
|
|
|
|
def test_bootstrap_creates_workspace(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "bootstrap_only.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=60)
|
|
assert result["success"]
|
|
assert file.dir_exists("/home/vulcan/workspace")
|
|
|
|
def test_bootstrap_creates_symlink(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "bootstrap_only.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=60)
|
|
assert result["success"]
|
|
assert os.path.islink("/workspace")
|
|
|
|
def test_bootstrap_generates_bootstrap_script(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "bootstrap_only.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=60)
|
|
assert result["success"]
|
|
assert file.file_exists("/bootstrap.sh")
|
|
|
|
def test_bootstrap_script_permissions(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "bootstrap_only.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=60)
|
|
assert result["success"]
|
|
mode = file.file_permissions("/bootstrap.sh")
|
|
assert mode & 0o111, f"/bootstrap.sh not executable, mode: {oct(mode)}"
|
|
|
|
def test_bootstrap_script_contains_user_creation(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "bootstrap_only.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=60)
|
|
assert result["success"]
|
|
assert file.file_exists("/bootstrap.sh")
|
|
assert file.file_contains("/bootstrap.sh", "useradd")
|
|
|
|
def test_bootstrap_script_contains_workspace_setup(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "bootstrap_only.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=60)
|
|
assert result["success"]
|
|
assert file.file_contains("/bootstrap.sh", "mkdir -p /home/vulcan/workspace")
|
|
|
|
def test_bootstrap_with_pacman_creates_sudoers(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "with_pacman.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=120)
|
|
assert result["success"], f"prebake failed: {result['stderr']}"
|
|
assert file.file_exists("/etc/sudoers.d/workshop")
|
|
|
|
def test_custom_bootstrap_executes(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "with_custom_bootstrap.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=60)
|
|
assert result["success"], f"Custom bootstrap failed: {result['stderr']}"
|
|
assert file.dir_exists("/custom-workspace")
|
|
|
|
|
|
@pytest.mark.prebake
|
|
class TestPrebakeDryRun:
|
|
|
|
def test_dry_run_does_not_create_user(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "bootstrap_only.yml")
|
|
env = {"HBW_DRY_RUN": "true"}
|
|
result = run_baker("prebake", config, cwd=work_dir, env=env, timeout=60)
|
|
assert result["success"], f"Dry-run failed: {result['stderr']}"
|
|
|
|
def test_dry_run_logs_script_content(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "bootstrap_only.yml")
|
|
env = {"HBW_DRY_RUN": "true", "RUST_LOG": "info"}
|
|
result = run_baker("prebake", config, cwd=work_dir, env=env, timeout=60)
|
|
assert result["success"]
|
|
combined = result["stdout"] + result["stderr"]
|
|
assert "DRY_RUN" in combined or "dry_run" in combined.lower()
|
|
|
|
|
|
@pytest.mark.prebake
|
|
class TestPrebakeHooks:
|
|
|
|
#@pytest.mark.xfail(reason="Hook execution fails with Permission denied in container environment", strict=False)
|
|
def test_early_hook_executes(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "with_hooks.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=60)
|
|
assert result["success"], f"prebake with hooks failed: {result['stderr']}"
|
|
combined = result["stdout"] + result["stderr"]
|
|
assert "early hook executed" in combined
|
|
|
|
#@pytest.mark.xfail(reason="Hook execution fails with Permission denied in container environment", strict=False)
|
|
def test_late_hook_executes(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "with_hooks.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=60)
|
|
assert result["success"], f"prebake with hooks failed: {result['stderr']}"
|
|
combined = result["stdout"] + result["stderr"]
|
|
assert "late hook executed" in combined
|
|
|
|
#@pytest.mark.xfail(reason="Hook execution fails with Permission denied in container environment", strict=False)
|
|
def test_hooks_run_in_order(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "with_hooks.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=60)
|
|
assert result["success"]
|
|
combined = result["stdout"] + result["stderr"]
|
|
early_pos = combined.find("early hook executed")
|
|
late_pos = combined.find("late hook executed")
|
|
assert early_pos != -1 and late_pos != -1
|
|
assert early_pos < late_pos, "Early hook should execute before late hook"
|
|
|
|
def test_empty_hooks_succeed(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "minimal.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=60)
|
|
assert result["success"], f"prebake with no hooks failed: {result['stderr']}"
|
|
|
|
|
|
@pytest.mark.prebake
|
|
class TestPrebakeDepsSystem:
|
|
|
|
@pytest.mark.xfail(reason="Package installation may fail in test environment", strict=False)
|
|
def test_pacman_packages_installed(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "with_pacman.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=120)
|
|
assert result["success"], f"prebake failed: {result['stderr']}"
|
|
assert package.package_installed("curl")
|
|
assert package.package_installed("wget")
|
|
|
|
def test_pacman_mirror_change(self, run_baker, work_dir):
|
|
mirror_config = os.path.join(work_dir, "mirror.yml")
|
|
with open(mirror_config, "w") as f:
|
|
f.write('''
|
|
version: "1.0"
|
|
environment:
|
|
builder: "baremetal"
|
|
bootstrap:
|
|
user: "vulcan"
|
|
dependencies:
|
|
system:
|
|
pacman:
|
|
packages: ["which"]
|
|
mirror: "https://mirrors.tuna.tsinghua.edu.cn/archlinux/$repo/os/$arch"
|
|
''')
|
|
result = run_baker("prebake", mirror_config, cwd=work_dir, timeout=120)
|
|
assert result["success"], f"prebake with mirror failed: {result['stderr']}"
|
|
|
|
|
|
@pytest.mark.prebake
|
|
class TestPrebakeEnvVars:
|
|
|
|
def test_prebake_envvars_injected(self, run_baker, work_dir):
|
|
hook_config = os.path.join(work_dir, "envvar_hook.yml")
|
|
with open(hook_config, "w") as f:
|
|
f.write('''
|
|
version: "1.0"
|
|
environment:
|
|
builder: "baremetal"
|
|
bootstrap:
|
|
user: "vulcan"
|
|
envvars:
|
|
prebake:
|
|
TEST_INJECTED: "injected_value"
|
|
hooks:
|
|
early:
|
|
- name: "check-env"
|
|
command: "echo $TEST_INJECTED"
|
|
''')
|
|
result = run_baker("prebake", hook_config, cwd=work_dir, timeout=60)
|
|
assert result["success"], f"prebake with envvars failed: {result['stderr']}"
|
|
combined = result["stdout"] + result["stderr"]
|
|
assert "injected_value" in combined
|
|
|
|
|
|
@pytest.mark.prebake
|
|
class TestPrebakeSecurity:
|
|
|
|
#@pytest.mark.xfail(reason="Privilege drop may fail in container without proper setup", strict=False)
|
|
def test_privilege_drop_after_bootstrap(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "with_security.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=60)
|
|
assert result["success"], f"prebake with security failed: {result['stderr']}"
|
|
|
|
|
|
@pytest.mark.prebake
|
|
class TestPrebakeStatus:
|
|
|
|
def test_status_files_written(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "bootstrap_only.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=60)
|
|
assert result["success"]
|
|
assert os.path.exists("/tmp/.hbwstatus")
|
|
|
|
def test_status_contains_bootstrap_stage(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "bootstrap_only.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=60)
|
|
assert result["success"]
|
|
with open("/tmp/.hbwstatus", "r") as f:
|
|
lines = f.readlines()
|
|
assert len(lines) > 0
|
|
bootstrap_lines = [l for l in lines if json.loads(l).get("name") == "bootstrap"]
|
|
assert len(bootstrap_lines) > 0, "No bootstrap stage found in status file"
|
|
data = json.loads(bootstrap_lines[-1])
|
|
assert data.get("phase") == "Prebake"
|
|
assert data.get("result") == "Success"
|
|
|
|
def test_status_contains_duration(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "bootstrap_only.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=60)
|
|
assert result["success"]
|
|
with open("/tmp/.hbwstatus", "r") as f:
|
|
data = json.loads(f.readline())
|
|
assert "duration_ms" in data
|
|
assert isinstance(data["duration_ms"], int)
|
|
|
|
def test_status_contains_timestamps(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "bootstrap_only.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=60)
|
|
assert result["success"]
|
|
with open("/tmp/.hbwstatus", "r") as f:
|
|
data = json.loads(f.readline())
|
|
assert "started_at" in data
|
|
assert "finished_at" in data
|
|
|
|
def test_status_written_for_each_stage(self, run_baker, work_dir):
|
|
config = os.path.join(FIXTURES, "with_pacman.yml")
|
|
result = run_baker("prebake", config, cwd=work_dir, timeout=120)
|
|
assert result["success"], f"prebake failed: {result['stderr']}"
|
|
with open("/tmp/.hbwstatus", "r") as f:
|
|
lines = f.readlines()
|
|
names = [json.loads(line).get("name") for line in lines]
|
|
assert "bootstrap" in names
|
|
assert "depssystem" in names
|