generated from coulomb/repo-seed
tegwick
8d509fc6f1
Port e2e-framework schema, runner, and reporter into wise-validator with sand-boxer CLI integration, validate run CLI, unit tests, registry capability, and operator docs.
104 lines
3.2 KiB
Python
104 lines
3.2 KiB
Python
"""Validation runner tests with mocked sand-boxer and SSH."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from wisevalidator.runner import run_validation
|
|
from wisevalidator.sandbox_client import SandboxHandle
|
|
|
|
FIXTURE_REPO = Path(__file__).parent / "fixtures" / "minimal_repo"
|
|
|
|
|
|
def test_run_validation_success() -> None:
|
|
handle = SandboxHandle(
|
|
sandbox_id="run12345",
|
|
host="coulombcore",
|
|
ssh="root@coulombcore",
|
|
remote_dir="/tmp/sandboxer/run12345",
|
|
compose_project="sbx-e2e-run12345",
|
|
raw={},
|
|
)
|
|
|
|
class FakeSSH:
|
|
remote_dir = handle.remote_dir
|
|
|
|
def wait_for_url(self, url: str, timeout: int = 120, interval: int = 5) -> bool:
|
|
return True
|
|
|
|
def run(self, cmd: str, *, timeout: int = 60) -> tuple[int, str]:
|
|
return 0, "test output\n"
|
|
|
|
with (
|
|
patch("wisevalidator.runner.create_sandbox", return_value=handle),
|
|
patch("wisevalidator.runner.destroy_sandbox") as destroy,
|
|
patch("wisevalidator.runner.SSHSession.from_reachability", return_value=FakeSSH()),
|
|
):
|
|
result = run_validation(FIXTURE_REPO, host="coulombcore")
|
|
|
|
assert result.passed is True
|
|
assert result.exit_code == 0
|
|
assert result.sandbox_id == "run12345"
|
|
destroy.assert_called_once_with("run12345")
|
|
|
|
|
|
def test_run_validation_keep_skips_destroy() -> None:
|
|
handle = SandboxHandle(
|
|
sandbox_id="keep1234",
|
|
host="coulombcore",
|
|
ssh="coulombcore",
|
|
remote_dir="/tmp/sandboxer/keep1234",
|
|
compose_project="sbx-e2e-keep1234",
|
|
raw={},
|
|
)
|
|
|
|
class FakeSSH:
|
|
remote_dir = handle.remote_dir
|
|
|
|
def wait_for_url(self, url: str, timeout: int = 120, interval: int = 5) -> bool:
|
|
return True
|
|
|
|
def run(self, cmd: str, *, timeout: int = 60) -> tuple[int, str]:
|
|
return 0, ""
|
|
|
|
with (
|
|
patch("wisevalidator.runner.create_sandbox", return_value=handle),
|
|
patch("wisevalidator.runner.destroy_sandbox") as destroy,
|
|
patch("wisevalidator.runner.SSHSession.from_reachability", return_value=FakeSSH()),
|
|
):
|
|
result = run_validation(FIXTURE_REPO, keep=True)
|
|
|
|
assert result.passed is True
|
|
destroy.assert_not_called()
|
|
|
|
|
|
def test_run_validation_health_failure() -> None:
|
|
handle = SandboxHandle(
|
|
sandbox_id="fail1234",
|
|
host="coulombcore",
|
|
ssh="coulombcore",
|
|
remote_dir="/tmp/sandboxer/fail1234",
|
|
compose_project=None,
|
|
raw={},
|
|
)
|
|
|
|
class FakeSSH:
|
|
remote_dir = handle.remote_dir
|
|
|
|
def wait_for_url(self, url: str, timeout: int = 120, interval: int = 5) -> bool:
|
|
return False
|
|
|
|
def run(self, cmd: str, *, timeout: int = 60) -> tuple[int, str]:
|
|
return 0, ""
|
|
|
|
with (
|
|
patch("wisevalidator.runner.create_sandbox", return_value=handle),
|
|
patch("wisevalidator.runner.destroy_sandbox") as destroy,
|
|
patch("wisevalidator.runner.SSHSession.from_reachability", return_value=FakeSSH()),
|
|
):
|
|
result = run_validation(FIXTURE_REPO)
|
|
|
|
assert result.passed is False
|
|
assert "Health check failed" in result.error
|
|
destroy.assert_called_once_with("fail1234") |