Simone
456db0a3e2
Windows produced an intermediate C:\CI\output\artifacts.zip that upload-artifact then re-zipped (zip-in-zip), while Linux deposited raw files. Make both paths consistent: the build stages raw outputs into a collect dir; collect zips it only as a transport archive, fetches, and extracts into the host artifact dir; actions/upload-artifact produces the single final archive. - _windows_build: param artifact_zip -> output_dir; stage raw files (in-place when source IS the collect dir, else wipe+copy), no Compress-Archive; absolute artifact-source resolved. - _windows_collect: zip guest dir as transport, fetch, extract (mirror _linux_collect); guest_path is now the collect directory. - job.py / build_run CLI: pass output_dir; rename --guest-artifact-zip -> --guest-output-dir. - Tests updated for the raw-file contract. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
463 lines
16 KiB
Python
463 lines
16 KiB
Python
"""Tests for ``ci_orchestrator job`` (Phase A4)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import click
|
|
import pytest
|
|
from click.testing import CliRunner
|
|
|
|
import ci_orchestrator.commands.job as job_module
|
|
from ci_orchestrator.__main__ import cli
|
|
from ci_orchestrator.backends.errors import BackendOperationFailed
|
|
from ci_orchestrator.backends.protocol import VmHandle
|
|
|
|
# ──────────────────────────────────────────────────────────── fakes
|
|
|
|
|
|
class _FakeBackend:
|
|
"""Records every backend call and creates the VMX file on clone."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
clone_fail: bool = False,
|
|
start_fail: bool = False,
|
|
delete_fail: bool = False,
|
|
ip: str | None = "10.99.0.42",
|
|
running_after: int = 0,
|
|
clone_guest_os: str = "ubuntu-64",
|
|
) -> None:
|
|
self.clone_fail = clone_fail
|
|
self.start_fail = start_fail
|
|
self.delete_fail = delete_fail
|
|
self.ip = ip
|
|
self.running_after = running_after
|
|
self.clone_guest_os = clone_guest_os
|
|
self.calls: list[str] = []
|
|
self._is_running_count = 0
|
|
|
|
def clone_linked(
|
|
self,
|
|
template: str,
|
|
snapshot: str,
|
|
name: str,
|
|
destination: str | None = None,
|
|
) -> VmHandle:
|
|
self.calls.append("clone")
|
|
if self.clone_fail:
|
|
raise BackendOperationFailed("clone", 1, "boom")
|
|
assert destination is not None
|
|
dst = Path(destination)
|
|
dst.parent.mkdir(parents=True, exist_ok=True)
|
|
dst.write_text(
|
|
f'config.version = "8"\nguestOS = "{self.clone_guest_os}"\n',
|
|
encoding="utf-8",
|
|
)
|
|
return VmHandle(identifier=str(dst), name=name)
|
|
|
|
def start(self, handle: VmHandle, headless: bool = True) -> None:
|
|
self.calls.append("start")
|
|
if self.start_fail:
|
|
raise BackendOperationFailed("start", 1, "no")
|
|
|
|
def stop(self, handle: VmHandle, hard: bool = False) -> None:
|
|
self.calls.append(f"stop:{'hard' if hard else 'soft'}")
|
|
|
|
def delete(self, handle: VmHandle) -> None:
|
|
self.calls.append("delete")
|
|
if self.delete_fail:
|
|
raise BackendOperationFailed("deleteVM", 1, "locked")
|
|
|
|
def is_running(self, handle: VmHandle) -> bool:
|
|
self._is_running_count += 1
|
|
return self._is_running_count > self.running_after
|
|
|
|
def get_ip(self, handle: VmHandle, timeout: float = 0.0) -> str | None:
|
|
return self.ip
|
|
|
|
def list_snapshots(self, handle: VmHandle) -> list[str]:
|
|
return []
|
|
|
|
|
|
def _patch_common(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
backend: _FakeBackend,
|
|
*,
|
|
transport_ready: bool = True,
|
|
linux_build_raises: bool = False,
|
|
) -> dict[str, list[Any]]:
|
|
"""Patch all in-process collaborators of the ``job`` command."""
|
|
calls: dict[str, list[Any]] = {
|
|
"linux_build": [],
|
|
"windows_build": [],
|
|
"linux_collect": [],
|
|
"windows_collect": [],
|
|
"manifest": [],
|
|
}
|
|
|
|
monkeypatch.setattr(job_module, "load_backend", lambda _cfg: backend)
|
|
|
|
class _ProbeOk:
|
|
def __enter__(self) -> _ProbeOk:
|
|
return self
|
|
|
|
def __exit__(self, *_e: object) -> None:
|
|
return None
|
|
|
|
def is_ready(self) -> bool:
|
|
return transport_ready
|
|
|
|
class _Store:
|
|
def get(self, _t: str) -> Any:
|
|
from ci_orchestrator.credentials import Credential
|
|
|
|
return Credential("ci", "pwd")
|
|
|
|
monkeypatch.setattr(job_module, "WinRmTransport", lambda *a, **k: _ProbeOk())
|
|
monkeypatch.setattr(job_module, "SshTransport", lambda *a, **k: _ProbeOk())
|
|
monkeypatch.setattr(job_module, "KeyringCredentialStore", lambda: _Store())
|
|
|
|
def _linux_build(**kwargs: Any) -> None:
|
|
calls["linux_build"].append(kwargs)
|
|
if linux_build_raises:
|
|
raise RuntimeError("build exploded")
|
|
|
|
def _windows_build(**kwargs: Any) -> None:
|
|
calls["windows_build"].append(kwargs)
|
|
|
|
def _linux_collect(**kwargs: Any) -> None:
|
|
calls["linux_collect"].append(kwargs)
|
|
# Simulate a transferred file so manifest writing has something to do.
|
|
Path(kwargs["host_dir"], "out.txt").write_text("ok", encoding="utf-8")
|
|
|
|
def _windows_collect(**kwargs: Any) -> None:
|
|
calls["windows_collect"].append(kwargs)
|
|
|
|
def _manifest(host_dir: Path, job_id: str, commit: str) -> int:
|
|
calls["manifest"].append((str(host_dir), job_id, commit))
|
|
return 1
|
|
|
|
monkeypatch.setattr(job_module, "_linux_build", _linux_build)
|
|
monkeypatch.setattr(job_module, "_windows_build", _windows_build)
|
|
monkeypatch.setattr(job_module, "_linux_collect", _linux_collect)
|
|
monkeypatch.setattr(job_module, "_windows_collect", _windows_collect)
|
|
monkeypatch.setattr(job_module, "_write_manifest", _manifest)
|
|
# Make poll loops instant.
|
|
monkeypatch.setattr(job_module.time, "sleep", lambda _s: None)
|
|
return calls
|
|
|
|
|
|
@pytest.fixture
|
|
def template_vmx(tmp_path: Path) -> Path:
|
|
"""Linux-flavoured VMX so auto-detect picks the SSH transport."""
|
|
vmx = tmp_path / "templates" / "LinuxBuild2404.vmx"
|
|
vmx.parent.mkdir(parents=True)
|
|
vmx.write_text('guestOS = "ubuntu-64"\n', encoding="utf-8")
|
|
return vmx
|
|
|
|
|
|
def _common_args(
|
|
template: Path,
|
|
*,
|
|
clone_base: Path,
|
|
artifact_base: Path,
|
|
extra: list[str] | None = None,
|
|
) -> list[str]:
|
|
return [
|
|
"job",
|
|
"--job-id",
|
|
"ci-test-1",
|
|
"--repo-url",
|
|
"http://gitea.local/org/repo.git",
|
|
"--branch",
|
|
"main",
|
|
"--template-path",
|
|
str(template),
|
|
"--clone-base-dir",
|
|
str(clone_base),
|
|
"--artifact-base-dir",
|
|
str(artifact_base),
|
|
"--ready-timeout",
|
|
"30",
|
|
"--poll-interval",
|
|
"1",
|
|
*(extra or []),
|
|
]
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────── tests
|
|
|
|
|
|
def test_job_happy_path_linux(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path, template_vmx: Path
|
|
) -> None:
|
|
backend = _FakeBackend(running_after=1)
|
|
calls = _patch_common(monkeypatch, backend)
|
|
|
|
base_clone = tmp_path / "build-vms"
|
|
base_artifact = tmp_path / "artifacts"
|
|
|
|
result = CliRunner().invoke(
|
|
cli, _common_args(template_vmx, clone_base=base_clone, artifact_base=base_artifact)
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
# Backend received the full lifecycle including delete in finally.
|
|
assert backend.calls[0] == "clone"
|
|
assert "start" in backend.calls
|
|
assert backend.calls[-1] == "delete"
|
|
# Linux build branch was taken (auto-detected from VMX guestOS).
|
|
assert len(calls["linux_build"]) == 1
|
|
assert calls["linux_build"][0]["build_command"] == ""
|
|
# Artifacts were collected into the per-job directory and manifest written.
|
|
assert (base_artifact / "ci-test-1" / "out.txt").is_file()
|
|
assert calls["manifest"] and calls["manifest"][0][1] == "ci-test-1"
|
|
# Clone directory removed.
|
|
assert not any((base_clone).iterdir()) or all(not p.exists() for p in base_clone.iterdir())
|
|
|
|
|
|
def test_job_skip_artifact_branch(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path, template_vmx: Path
|
|
) -> None:
|
|
backend = _FakeBackend(running_after=0)
|
|
calls = _patch_common(monkeypatch, backend)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
_common_args(
|
|
template_vmx,
|
|
clone_base=tmp_path / "vms",
|
|
artifact_base=tmp_path / "art",
|
|
extra=["--skip-artifact"],
|
|
),
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert calls["linux_collect"] == []
|
|
assert calls["windows_collect"] == []
|
|
assert calls["manifest"] == []
|
|
assert backend.calls[-1] == "delete"
|
|
|
|
|
|
def test_job_cleanup_on_build_failure(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path, template_vmx: Path
|
|
) -> None:
|
|
"""If the build raises, the VM must still be destroyed (try/finally)."""
|
|
backend = _FakeBackend(running_after=0)
|
|
_patch_common(monkeypatch, backend, linux_build_raises=True)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
_common_args(template_vmx, clone_base=tmp_path / "vms", artifact_base=tmp_path / "art"),
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "build exploded" in result.output
|
|
assert "delete" in backend.calls
|
|
# Stop attempt happened before delete.
|
|
assert any(c.startswith("stop:") for c in backend.calls)
|
|
|
|
|
|
def test_job_cleanup_on_clone_failure(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path, template_vmx: Path
|
|
) -> None:
|
|
backend = _FakeBackend(clone_fail=True)
|
|
_patch_common(monkeypatch, backend)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
_common_args(template_vmx, clone_base=tmp_path / "vms", artifact_base=tmp_path / "art"),
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "clone failed" in result.output
|
|
# Backend.start/delete must not have been called: handle is None.
|
|
assert backend.calls == ["clone"]
|
|
|
|
|
|
def test_job_rejects_missing_template(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
backend = _FakeBackend()
|
|
_patch_common(monkeypatch, backend)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
_common_args(
|
|
tmp_path / "ghost.vmx",
|
|
clone_base=tmp_path / "vms",
|
|
artifact_base=tmp_path / "art",
|
|
),
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "Template VMX not found" in result.output
|
|
assert backend.calls == []
|
|
|
|
|
|
def test_job_windows_branch_with_overrides(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
"""Windows-flavoured VMX selects the WinRM build path; CPU/RAM override applied."""
|
|
vmx = tmp_path / "templates" / "WinBuild2025.vmx"
|
|
vmx.parent.mkdir(parents=True)
|
|
vmx.write_text('guestOS = "windows10srv-64"\n', encoding="utf-8")
|
|
|
|
backend = _FakeBackend(running_after=0, clone_guest_os="windows10srv-64")
|
|
calls = _patch_common(monkeypatch, backend)
|
|
extra = [
|
|
"--guest-cpu",
|
|
"4",
|
|
"--guest-memory-mb",
|
|
"8192",
|
|
"--extra-env-json",
|
|
'{"BUILD_TAG":"abc"}',
|
|
]
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
_common_args(
|
|
vmx,
|
|
clone_base=tmp_path / "vms",
|
|
artifact_base=tmp_path / "art",
|
|
extra=extra,
|
|
),
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert calls["windows_build"], "Windows build branch not taken"
|
|
assert calls["windows_build"][0]["extra_env"] == {"BUILD_TAG": "abc"}
|
|
# CPU/RAM override patched the clone VMX.
|
|
clone_vmx = next((tmp_path / "vms").rglob("*.vmx"), None)
|
|
# After successful job the clone dir is deleted, so the file is gone.
|
|
# Check that the Windows collect helper got the collect dir instead.
|
|
assert calls["windows_collect"][0]["guest_path"] == "C:\\CI\\output"
|
|
assert clone_vmx is None # cleaned up
|
|
assert backend.calls[-1] == "delete"
|
|
|
|
|
|
def test_job_invalid_extra_env_json(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path, template_vmx: Path
|
|
) -> None:
|
|
backend = _FakeBackend()
|
|
_patch_common(monkeypatch, backend)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
_common_args(
|
|
template_vmx,
|
|
clone_base=tmp_path / "vms",
|
|
artifact_base=tmp_path / "art",
|
|
extra=["--extra-env-json", '{"bad-key":"x"}'],
|
|
),
|
|
)
|
|
assert result.exit_code != 0
|
|
assert "not a valid env var name" in result.output
|
|
# Failure is parameter-time; clone never started.
|
|
assert backend.calls == []
|
|
|
|
|
|
# ──────────────────────────────────────────────── helper-level tests
|
|
|
|
|
|
def test_apply_vmx_overrides_replaces_existing_lines(tmp_path: Path) -> None:
|
|
vmx = tmp_path / "x.vmx"
|
|
vmx.write_text(
|
|
'config.version = "8"\nnumvcpus = "1"\nmemsize = "2048"\n',
|
|
encoding="utf-8",
|
|
)
|
|
job_module._apply_vmx_overrides(vmx, cpu=8, memory_mb=16384)
|
|
text = vmx.read_text(encoding="utf-8")
|
|
assert 'numvcpus = "8"' in text
|
|
assert 'memsize = "16384"' in text
|
|
|
|
|
|
def test_apply_vmx_overrides_appends_when_missing(tmp_path: Path) -> None:
|
|
vmx = tmp_path / "x.vmx"
|
|
vmx.write_text('config.version = "8"\n', encoding="utf-8")
|
|
job_module._apply_vmx_overrides(vmx, cpu=2, memory_mb=4096)
|
|
text = vmx.read_text(encoding="utf-8")
|
|
assert 'numvcpus = "2"' in text
|
|
assert 'memsize = "4096"' in text
|
|
|
|
|
|
def test_read_guest_os_detects_linux(tmp_path: Path) -> None:
|
|
vmx = tmp_path / "x.vmx"
|
|
vmx.write_text('guestOS = "ubuntu-64"\n', encoding="utf-8")
|
|
assert job_module._read_guest_os_from_vmx(vmx) == "linux"
|
|
|
|
|
|
def test_read_guest_os_defaults_to_windows(tmp_path: Path) -> None:
|
|
vmx = tmp_path / "x.vmx"
|
|
vmx.write_text("# nothing useful\n", encoding="utf-8")
|
|
assert job_module._read_guest_os_from_vmx(vmx) == "windows"
|
|
|
|
|
|
def test_parse_extra_env_json_empty_inputs() -> None:
|
|
assert job_module._parse_extra_env_json("") == {}
|
|
assert job_module._parse_extra_env_json("{}") == {}
|
|
|
|
|
|
def test_parse_extra_env_json_rejects_non_object() -> None:
|
|
with pytest.raises(click.exceptions.UsageError):
|
|
job_module._parse_extra_env_json("[]")
|
|
|
|
|
|
def test_read_guest_os_handles_oserror(tmp_path: Path) -> None:
|
|
"""Unreadable VMX (e.g. a directory) returns the safe ``windows`` default."""
|
|
p = tmp_path / "is_a_dir.vmx"
|
|
p.mkdir()
|
|
assert job_module._read_guest_os_from_vmx(p) == "windows"
|
|
|
|
|
|
def test_apply_vmx_overrides_noop_when_both_zero(tmp_path: Path) -> None:
|
|
vmx = tmp_path / "x.vmx"
|
|
vmx.write_text('numvcpus = "1"\n', encoding="utf-8")
|
|
job_module._apply_vmx_overrides(vmx, cpu=0, memory_mb=0)
|
|
assert vmx.read_text(encoding="utf-8") == 'numvcpus = "1"\n'
|
|
|
|
|
|
def test_parse_extra_env_json_invalid_json_raises() -> None:
|
|
with pytest.raises(click.exceptions.UsageError):
|
|
job_module._parse_extra_env_json("{not json")
|
|
|
|
|
|
def test_parse_extra_env_json_coerces_value_types() -> None:
|
|
out = job_module._parse_extra_env_json('{"A":null,"B":42,"C":"x"}')
|
|
assert out == {"A": "", "B": "42", "C": "x"}
|
|
|
|
|
|
def test_parse_extra_env_json_rejects_non_string_key() -> None:
|
|
# JSON keys are always strings, so coerce via raw call to guarantee branch.
|
|
with pytest.raises(click.exceptions.UsageError):
|
|
job_module._parse_extra_env_json('{"1bad":"x"}')
|
|
|
|
|
|
def test_job_wait_running_swallows_backend_error(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from ci_orchestrator.backends.errors import BackendError
|
|
from ci_orchestrator.backends.protocol import VmHandle
|
|
|
|
class _Boom:
|
|
def is_running(self, _h: VmHandle) -> bool:
|
|
raise BackendError("nope")
|
|
|
|
monkeypatch.setattr(job_module.time, "sleep", lambda _s: None)
|
|
seq = iter([0.0, 1.0])
|
|
monkeypatch.setattr(job_module, "_now", lambda: next(seq))
|
|
assert (
|
|
job_module._wait_running(_Boom(), VmHandle("x"), deadline=0.5, poll=0.001) # type: ignore[arg-type]
|
|
is False
|
|
)
|
|
|
|
|
|
def test_job_wait_for_ip_returns_none_on_backend_error(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from ci_orchestrator.backends.errors import BackendError
|
|
from ci_orchestrator.backends.protocol import VmHandle
|
|
|
|
class _Boom:
|
|
def get_ip(self, _h: VmHandle, timeout: float = 0.0) -> str | None:
|
|
raise BackendError("ip nope")
|
|
|
|
monkeypatch.setattr(job_module.time, "sleep", lambda _s: None)
|
|
seq = iter([0.0, 1.0])
|
|
monkeypatch.setattr(job_module, "_now", lambda: next(seq))
|
|
assert (
|
|
job_module._wait_for_ip(_Boom(), VmHandle("x"), deadline=0.5, poll=0.001) # type: ignore[arg-type]
|
|
is None
|
|
)
|