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
No EOL
3.2 KiB
Python
104 lines
No EOL
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") |