feat(a3): port build pipeline (vm new, build run, artifacts collect)

Implements Phase A3 of plans/implementation-plan-A-B.md:

- commands/vm.py: vm new (replaces New-BuildVM.ps1)

- commands/build.py: build run (replaces Invoke-RemoteBuild.ps1) with WinRM/SSH dispatch and stdout streaming for act_runner

- commands/artifacts.py: artifacts collect (replaces Get-BuildArtifacts.ps1) using transport.fetch()

- 3 PS scripts reduced to shims preserving \0

- Pester tests/{New-BuildVM,Wait-VMReady,Remove-BuildVM}.Tests.ps1 removed; equivalent cases covered in pytest

- Phase C hook: build/artifacts accept opaque VmHandle/vmx; no Windows path assumptions

Tests: 91 pytest, ruff clean, mypy --strict clean, coverage 78.27% (>=70 gate).

Master checklist + step A3 attivita + 'definizione di fatto' updated; A3-closeout.md added.
This commit is contained in:
2026-05-14 17:20:34 +02:00
parent 096ba7fe16
commit 816a15503e
15 changed files with 1900 additions and 1061 deletions
-106
View File
@@ -1,106 +0,0 @@
#Requires -Version 5.1
<#
.SYNOPSIS
Pester v5 tests for scripts/New-BuildVM.ps1
#>
BeforeAll {
$script:ScriptPath = Join-Path $PSScriptRoot '..\scripts\New-BuildVM.ps1'
$script:CloneBaseDir = Join-Path $env:TEMP "CI-Tests-CloneBase-$PID"
New-Item -ItemType Directory -Path $script:CloneBaseDir -Force | Out-Null
# Fake vmrun.ps1 — exits with $env:FAKE_VMRUN_EXIT (default 0)
# On success it also creates the destination VMX so the post-clone Test-Path passes.
# Using .ps1 instead of .cmd avoids Windows cmd redirect limitations inside if blocks.
$script:FakeVmrun = Join-Path $env:TEMP "fake-vmrun-$PID.ps1"
Set-Content $script:FakeVmrun -Value @'
# Args: -T ws clone <template> <dst_vmx> linked -snapshot <snapshot>
$exitCode = if ($env:FAKE_VMRUN_EXIT) { [int]$env:FAKE_VMRUN_EXIT } else { 0 }
if ($exitCode -eq 0) {
# Create the destination VMX (5th positional arg) so Test-Path in caller passes.
# Real vmrun creates the clone dir+file; we must do the same here.
$dstVmx = $args[4] # -T ws clone <tmpl> <dst> ...
$dstDir = Split-Path $dstVmx -Parent
if (-not (Test-Path $dstDir)) { New-Item -ItemType Directory -Path $dstDir -Force | Out-Null }
Set-Content $dstVmx ''
}
exit $exitCode
'@
# Fake template VMX — just needs to exist
$script:FakeTemplate = Join-Path $env:TEMP "fake-template-$PID.vmx"
Set-Content $script:FakeTemplate -Value 'config.version = "8"'
}
AfterAll {
Remove-Item $script:CloneBaseDir -Recurse -Force -ErrorAction SilentlyContinue
Remove-Item $script:FakeVmrun -Force -ErrorAction SilentlyContinue
Remove-Item $script:FakeTemplate -Force -ErrorAction SilentlyContinue
$env:FAKE_VMRUN_EXIT = $null
}
# ─────────────────────────────────────────────────────────────────────────────
Describe 'New-BuildVM — parameter validation' {
It 'throws when TemplatePath does not exist' {
{
& $script:ScriptPath `
-TemplatePath 'C:\DoesNotExist\template.vmx' `
-CloneBaseDir $script:CloneBaseDir `
-JobId 'test-job-1' `
-VmrunPath $script:FakeVmrun
} | Should -Throw
}
}
# ─────────────────────────────────────────────────────────────────────────────
Describe 'New-BuildVM — vmrun failure' {
It 'throws and cleans up clone dir when vmrun clone exits non-zero' {
$env:FAKE_VMRUN_EXIT = '1'
{
& $script:ScriptPath `
-TemplatePath $script:FakeTemplate `
-CloneBaseDir $script:CloneBaseDir `
-JobId 'test-job-fail' `
-VmrunPath $script:FakeVmrun
} | Should -Throw -ExpectedMessage '*clone failed*'
# No leftover clone dir
$leftover = Get-ChildItem $script:CloneBaseDir -Directory |
Where-Object { $_.Name -like '*test-job-fail*' }
$leftover | Should -BeNullOrEmpty
}
AfterEach { $env:FAKE_VMRUN_EXIT = $null }
}
# ─────────────────────────────────────────────────────────────────────────────
Describe 'New-BuildVM — clone name format' {
It 'clone folder is named Clone_{JobId}_{yyyyMMdd_HHmmss}' {
$env:FAKE_VMRUN_EXIT = '0'
$vmx = & $script:ScriptPath `
-TemplatePath $script:FakeTemplate `
-CloneBaseDir $script:CloneBaseDir `
-JobId 'run-42' `
-VmrunPath $script:FakeVmrun
$vmx | Should -Match 'Clone_run-42_\d{8}_\d{6}'
}
It 'returns a string ending in .vmx' {
$env:FAKE_VMRUN_EXIT = '0'
$vmx = & $script:ScriptPath `
-TemplatePath $script:FakeTemplate `
-CloneBaseDir $script:CloneBaseDir `
-JobId 'run-99' `
-VmrunPath $script:FakeVmrun
$vmx | Should -Match '\.vmx$'
}
AfterEach {
$env:FAKE_VMRUN_EXIT = $null
# Clean up clones created by successful tests
Get-ChildItem $script:CloneBaseDir -Directory -ErrorAction SilentlyContinue |
Remove-Item -Recurse -Force -ErrorAction SilentlyContinue
}
}
-68
View File
@@ -1,68 +0,0 @@
#Requires -Version 5.1
<#
.SYNOPSIS
Pester v5 tests for scripts/Remove-BuildVM.ps1
#>
BeforeAll {
$script:ScriptPath = Join-Path $PSScriptRoot '..\scripts\Remove-BuildVM.ps1'
# Fake vmrun: accepts any args, always exits 0 (best-effort ops don't need to fail here)
$script:FakeVmrun = Join-Path $env:TEMP "fake-vmrun-remove-$PID.cmd"
Set-Content $script:FakeVmrun -Value '@echo off & exit /b 0'
}
AfterAll {
Remove-Item $script:FakeVmrun -Force -ErrorAction SilentlyContinue
}
# ─────────────────────────────────────────────────────────────────────────────
Describe 'Remove-BuildVM — missing VMX' {
It 'returns without error when VMX does not exist' {
{ & $script:ScriptPath -VMPath 'C:\DoesNotExist\clone.vmx' -VmrunPath $script:FakeVmrun } |
Should -Not -Throw
}
It 'removes partial clone directory even when VMX is missing' {
$partialDir = Join-Path $env:TEMP "partial-clone-$PID"
New-Item -ItemType Directory -Path $partialDir -Force | Out-Null
$fakeMissingVmx = Join-Path $partialDir 'missing.vmx'
& $script:ScriptPath -VMPath $fakeMissingVmx -VmrunPath $script:FakeVmrun
Test-Path $partialDir | Should -Be $false
}
}
# ─────────────────────────────────────────────────────────────────────────────
Describe 'Remove-BuildVM — WhatIf' {
It 'does not remove clone directory when -WhatIf is passed' {
$cloneDir = Join-Path $env:TEMP "whatif-clone-$PID"
$cloneVmx = Join-Path $cloneDir 'whatif-clone.vmx'
New-Item -ItemType Directory -Path $cloneDir -Force | Out-Null
Set-Content $cloneVmx -Value 'config.version = "8"'
& $script:ScriptPath -VMPath $cloneVmx -VmrunPath $script:FakeVmrun -WhatIf
Test-Path $cloneDir | Should -Be $true
Remove-Item $cloneDir -Recurse -Force -ErrorAction SilentlyContinue
}
}
# ─────────────────────────────────────────────────────────────────────────────
Describe 'Remove-BuildVM — successful destroy' {
It 'removes the clone directory when VMX exists and vmrun succeeds' {
$cloneDir = Join-Path $env:TEMP "live-clone-$PID"
$cloneVmx = Join-Path $cloneDir 'live-clone.vmx'
New-Item -ItemType Directory -Path $cloneDir -Force | Out-Null
Set-Content $cloneVmx -Value 'config.version = "8"'
& $script:ScriptPath `
-VMPath $cloneVmx `
-VmrunPath $script:FakeVmrun `
-GracefulTimeoutSeconds 1 # short timeout for test speed
Test-Path $cloneDir | Should -Be $false
}
}
-92
View File
@@ -1,92 +0,0 @@
#Requires -Version 5.1
<#
.SYNOPSIS
Pester v5 tests for scripts/Wait-VMReady.ps1
#>
BeforeAll {
$script:ScriptPath = Join-Path $PSScriptRoot '..\scripts\Wait-VMReady.ps1'
# Fake vmrun: getGuestIPAddress exits with %FAKE_VMRUN_EXIT% (default 1 = not ready)
$script:FakeVmrun = Join-Path $env:TEMP "fake-vmrun-wait-$PID.cmd"
Set-Content $script:FakeVmrun -Value @'
@echo off
if "%FAKE_VMRUN_EXIT%"=="" ( exit /b 1 ) else ( exit /b %FAKE_VMRUN_EXIT% )
'@
# Fake VMX file — just needs to exist
$script:FakeVmx = Join-Path $env:TEMP "fake-wait-$PID.vmx"
Set-Content $script:FakeVmx -Value 'config.version = "8"'
}
AfterAll {
Remove-Item $script:FakeVmrun -Force -ErrorAction SilentlyContinue
Remove-Item $script:FakeVmx -Force -ErrorAction SilentlyContinue
$env:FAKE_VMRUN_EXIT = $null
}
# ─────────────────────────────────────────────────────────────────────────────
Describe 'Wait-VMReady — missing vmrun' {
It 'throws when VmrunPath does not exist' {
{
& $script:ScriptPath `
-VMPath $script:FakeVmx `
-IPAddress '192.168.79.101' `
-VmrunPath 'C:\DoesNotExist\vmrun.exe'
} | Should -Throw
}
}
# ─────────────────────────────────────────────────────────────────────────────
Describe 'Wait-VMReady — timeout' {
It 'throws with timeout message when VM never becomes ready' {
$env:FAKE_VMRUN_EXIT = '1' # vmrun getGuestIPAddress always fails → VM not running
{
& $script:ScriptPath `
-VMPath $script:FakeVmx `
-IPAddress '192.168.79.101' `
-VmrunPath $script:FakeVmrun `
-TimeoutSeconds 5 `
-PollIntervalSeconds 2 `
-SkipPing
} | Should -Throw -ExpectedMessage '*Timed out*'
}
AfterEach { $env:FAKE_VMRUN_EXIT = $null }
}
# ─────────────────────────────────────────────────────────────────────────────
Describe 'Wait-VMReady — IP validation' {
It 'rejects an invalid IP address' {
{
& $script:ScriptPath `
-VMPath $script:FakeVmx `
-IPAddress '999.999.999.999' `
-VmrunPath $script:FakeVmrun
} | Should -Throw
}
It 'accepts a valid IP address without immediate error from parameter binding' {
# This will still time out (vmrun fake exits 1), but the IP itself must not
# cause a parameter validation error — that would throw before the timeout.
$env:FAKE_VMRUN_EXIT = '1'
$err = $null
try {
& $script:ScriptPath `
-VMPath $script:FakeVmx `
-IPAddress '192.168.79.101' `
-VmrunPath $script:FakeVmrun `
-TimeoutSeconds 3 `
-PollIntervalSeconds 2 `
-SkipPing `
-ErrorAction Stop
} catch {
$err = $_
}
# Error must be the timeout, not a parameter validation error
"$err" | Should -Match 'Timed out'
"$err" | Should -Not -Match 'Cannot validate'
$env:FAKE_VMRUN_EXIT = $null
}
}
+289
View File
@@ -0,0 +1,289 @@
"""Tests for ``ci_orchestrator artifacts collect``."""
from __future__ import annotations
import json
import tarfile
from pathlib import Path
from typing import Any, ClassVar
import pytest
from click.testing import CliRunner
import ci_orchestrator.commands.artifacts as artifacts_module
from ci_orchestrator.__main__ import cli
from ci_orchestrator.credentials import Credential
from ci_orchestrator.transport.errors import TransportCommandError
class _FakeResult:
def __init__(self, stdout: str = "OK", returncode: int = 0) -> None:
self.stdout = stdout
self.stderr = ""
self.returncode = returncode
@property
def ok(self) -> bool:
return self.returncode == 0
class _FakeTransport:
instances: ClassVar[list[_FakeTransport]] = []
def __init__(self, *args: Any, **kwargs: Any) -> None:
self.runs: list[str] = []
self.fetched: list[tuple[str, str]] = []
self.fetch_payloads: dict[str, bytes] = {}
self.run_overrides: dict[str, _FakeResult] = {}
type(self).instances.append(self)
def __enter__(self) -> _FakeTransport:
return self
def __exit__(self, *_e: object) -> None:
return None
def run(self, script: str, *, check: bool = True) -> _FakeResult:
self.runs.append(script)
for key, override in self.run_overrides.items():
if key in script:
if check and not override.ok:
raise TransportCommandError(
override.returncode, override.stdout, override.stderr
)
return override
return _FakeResult()
def fetch(self, remote_path: str, local_path: str) -> None:
self.fetched.append((remote_path, str(local_path)))
payload = self.fetch_payloads.get(remote_path, b"data")
Path(local_path).write_bytes(payload)
@pytest.fixture(autouse=True)
def _reset_instances() -> None:
_FakeTransport.instances = []
def _patch(monkeypatch: pytest.MonkeyPatch) -> None:
class _Store:
def get(self, _t: str) -> Credential:
return Credential("ci", "pwd")
monkeypatch.setattr(artifacts_module, "WinRmTransport", _FakeTransport)
monkeypatch.setattr(artifacts_module, "SshTransport", _FakeTransport)
monkeypatch.setattr(artifacts_module, "KeyringCredentialStore", lambda: _Store())
# ── Windows happy path ────────────────────────────────────────────────────
def test_collect_windows_happy_path(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_patch(monkeypatch)
out_dir = tmp_path / "artifacts"
result = CliRunner().invoke(
cli,
[
"artifacts",
"collect",
"--ip-address",
"10.0.0.7",
"--guest-artifact-path",
"C:\\CI\\output\\artifacts.zip",
"--host-artifact-dir",
str(out_dir),
],
)
assert result.exit_code == 0, result.output
assert (out_dir / "artifacts.zip").is_file()
instance = _FakeTransport.instances[0]
assert any("Test-Path" in c for c in instance.runs)
assert instance.fetched == [
("C:\\CI\\output\\artifacts.zip", str(out_dir / "artifacts.zip"))
]
def test_collect_windows_artifact_missing(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_patch(monkeypatch)
class _Missing(_FakeTransport):
def __init__(self, *a: Any, **kw: Any) -> None:
super().__init__(*a, **kw)
self.run_overrides["Test-Path"] = _FakeResult(stdout="MISSING\r\n")
monkeypatch.setattr(artifacts_module, "WinRmTransport", _Missing)
result = CliRunner().invoke(
cli,
[
"artifacts",
"collect",
"--ip-address",
"10.0.0.7",
"--guest-artifact-path",
"C:\\CI\\output\\artifacts.zip",
"--host-artifact-dir",
str(tmp_path / "out"),
],
)
assert result.exit_code != 0
assert "artifact not found" in result.output
def test_collect_windows_include_logs(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_patch(monkeypatch)
class _WithLogs(_FakeTransport):
def __init__(self, *a: Any, **kw: Any) -> None:
super().__init__(*a, **kw)
self.run_overrides["Get-ChildItem"] = _FakeResult(
stdout="C:\\CI\\build\\one.log\r\nC:\\CI\\build\\sub\\two.log\r\n"
)
monkeypatch.setattr(artifacts_module, "WinRmTransport", _WithLogs)
result = CliRunner().invoke(
cli,
[
"artifacts",
"collect",
"--ip-address",
"10.0.0.7",
"--guest-artifact-path",
"C:\\CI\\output\\artifacts.zip",
"--host-artifact-dir",
str(tmp_path / "out"),
"--include-logs",
],
)
assert result.exit_code == 0, result.output
fetched_remotes = [r for r, _ in _FakeTransport.instances[0].fetched]
assert "C:\\CI\\build\\one.log" in fetched_remotes
assert "C:\\CI\\build\\sub\\two.log" in fetched_remotes
def test_collect_writes_manifest_when_job_id_set(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
_patch(monkeypatch)
out = tmp_path / "out"
result = CliRunner().invoke(
cli,
[
"artifacts",
"collect",
"--ip-address",
"10.0.0.7",
"--guest-artifact-path",
"C:\\CI\\output\\artifacts.zip",
"--host-artifact-dir",
str(out),
"--job-id",
"run-42",
"--commit",
"abc1234",
],
)
assert result.exit_code == 0, result.output
manifest_path = out / "manifest.json"
assert manifest_path.is_file()
data = json.loads(manifest_path.read_text(encoding="utf-8"))
assert data["jobId"] == "run-42"
assert data["commit"] == "abc1234"
names = [f["name"] for f in data["files"]]
assert "artifacts.zip" in names
# ── Linux happy path ──────────────────────────────────────────────────────
def test_collect_linux_extracts_tarball(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
_patch(monkeypatch)
# Build a real tarball that the fake transport will "fetch".
payload_dir = tmp_path / "payload"
payload_dir.mkdir()
(payload_dir / "binary.bin").write_bytes(b"\x01\x02\x03")
(payload_dir / "log.txt").write_text("hello", encoding="utf-8")
tar_local = tmp_path / "src.tar.gz"
with tarfile.open(tar_local, "w:gz") as tf:
for f in payload_dir.iterdir():
tf.add(f, arcname=f.name)
payload_bytes = tar_local.read_bytes()
class _LinuxTransport(_FakeTransport):
def __init__(self, *a: Any, **kw: Any) -> None:
super().__init__(*a, **kw)
# Any fetched path returns the same tarball payload.
self.fetch_payloads["__default__"] = payload_bytes
def fetch(self, remote_path: str, local_path: str) -> None:
self.fetched.append((remote_path, str(local_path)))
Path(local_path).write_bytes(payload_bytes)
monkeypatch.setattr(artifacts_module, "SshTransport", _LinuxTransport)
out = tmp_path / "out"
result = CliRunner().invoke(
cli,
[
"artifacts",
"collect",
"--ip-address",
"10.0.0.5",
"--guest-os",
"linux",
"--guest-artifact-path",
"/opt/ci/output",
"--host-artifact-dir",
str(out),
"--ssh-key-path",
str(tmp_path / "key"),
],
)
assert result.exit_code == 0, result.output
assert (out / "binary.bin").read_bytes() == b"\x01\x02\x03"
assert (out / "log.txt").read_text(encoding="utf-8") == "hello"
cmds = _FakeTransport.instances[0].runs
assert any("tar -czf" in c for c in cmds)
# Cleanup of the temp tarball on the guest was attempted.
assert any("rm -f" in c for c in cmds)
def test_collect_linux_no_files_raises(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
_patch(monkeypatch)
# Empty tarball => extraction yields nothing.
empty_tar = tmp_path / "empty.tar.gz"
with tarfile.open(empty_tar, "w:gz"):
pass
payload = empty_tar.read_bytes()
class _Empty(_FakeTransport):
def fetch(self, remote_path: str, local_path: str) -> None:
self.fetched.append((remote_path, str(local_path)))
Path(local_path).write_bytes(payload)
monkeypatch.setattr(artifacts_module, "SshTransport", _Empty)
result = CliRunner().invoke(
cli,
[
"artifacts",
"collect",
"--ip-address",
"10.0.0.5",
"--guest-os",
"linux",
"--guest-artifact-path",
"/opt/ci/output",
"--host-artifact-dir",
str(tmp_path / "out"),
"--ssh-key-path",
str(tmp_path / "key"),
],
)
assert result.exit_code != 0
assert "no files were transferred" in result.output
+364
View File
@@ -0,0 +1,364 @@
"""Tests for ``ci_orchestrator build run``."""
from __future__ import annotations
from pathlib import Path
from typing import Any, ClassVar
import pytest
from click.testing import CliRunner
import ci_orchestrator.commands.build as build_module
from ci_orchestrator.__main__ import cli
from ci_orchestrator.credentials import Credential
from ci_orchestrator.transport.errors import TransportCommandError
class _FakeResult:
def __init__(self, stdout: str = "", stderr: str = "", returncode: int = 0) -> None:
self.stdout = stdout
self.stderr = stderr
self.returncode = returncode
@property
def ok(self) -> bool:
return self.returncode == 0
class _FakeTransport:
"""Records every ``run/copy/fetch`` call."""
instances: ClassVar[list[_FakeTransport]] = []
def __init__(self, *args: Any, **kwargs: Any) -> None:
self.args = args
self.kwargs = kwargs
self.runs: list[tuple[str, bool]] = []
self.copies: list[tuple[str, str]] = []
self.fetches: list[tuple[str, str]] = []
self.run_results: dict[int, _FakeResult] = {}
self.default_result = _FakeResult(stdout="ok")
type(self).instances.append(self)
def __enter__(self) -> _FakeTransport:
return self
def __exit__(self, *_e: object) -> None:
return None
def run(self, script: str, *, check: bool = True) -> _FakeResult:
self.runs.append((script, check))
result = self.run_results.get(len(self.runs) - 1, self.default_result)
if check and not result.ok:
raise TransportCommandError(result.returncode, result.stdout, result.stderr)
return result
def copy(self, local_path: str, remote_path: str) -> None:
self.copies.append((str(local_path), remote_path))
def fetch(self, remote_path: str, local_path: str) -> None:
self.fetches.append((remote_path, str(local_path)))
@pytest.fixture(autouse=True)
def _reset_instances() -> None:
_FakeTransport.instances = []
def _patch(monkeypatch: pytest.MonkeyPatch) -> type[_FakeTransport]:
class _Store:
def get(self, _t: str) -> Credential:
return Credential("ci", "pwd")
monkeypatch.setattr(build_module, "WinRmTransport", _FakeTransport)
monkeypatch.setattr(build_module, "SshTransport", _FakeTransport)
monkeypatch.setattr(build_module, "KeyringCredentialStore", lambda: _Store())
return _FakeTransport
# ── parameter validation ───────────────────────────────────────────────────
def test_build_run_requires_source(monkeypatch: pytest.MonkeyPatch) -> None:
_patch(monkeypatch)
result = CliRunner().invoke(
cli, ["build", "run", "--ip-address", "10.0.0.1"]
)
assert result.exit_code != 0
assert "either --host-source-dir or --clone-url" in result.output
def test_build_run_rejects_both_modes(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_patch(monkeypatch)
src = tmp_path / "src"
src.mkdir()
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.1",
"--host-source-dir",
str(src),
"--clone-url",
"https://example/repo.git",
],
)
assert result.exit_code != 0
assert "not both" in result.output
def test_build_run_rejects_missing_source_dir(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_patch(monkeypatch)
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.1",
"--host-source-dir",
str(tmp_path / "ghost"),
],
)
assert result.exit_code != 0
assert "does not exist" in result.output
def test_build_run_rejects_bad_extra_env(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_patch(monkeypatch)
src = tmp_path / "src"
src.mkdir()
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.1",
"--guest-os",
"linux",
"--host-source-dir",
str(src),
"--build-command",
"make",
"--extra-env",
"NO_EQUALS",
],
)
assert result.exit_code != 0
assert "KEY=VALUE" in result.output
# ── happy path: Linux ──────────────────────────────────────────────────────
def test_build_run_linux_clone_url_happy_path(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
_patch(monkeypatch)
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.5",
"--guest-os",
"linux",
"--clone-url",
"https://example/repo.git",
"--clone-branch",
"feature/x",
"--clone-commit",
"deadbeef",
"--clone-submodules",
"--build-command",
"make all",
"--guest-linux-work-dir",
"/work",
"--guest-linux-output-dir",
"/out",
"--guest-artifact-source",
"build",
"--extra-env",
"FOO=bar baz",
"--extra-env",
"TOKEN=secret",
"--ssh-key-path",
str(tmp_path / "key"),
],
)
assert result.exit_code == 0, result.output
instance = _FakeTransport.instances[0]
cmds = [c[0] for c in instance.runs]
assert any("rm -rf /work && mkdir -p /work" in c for c in cmds)
assert any("git clone --depth 1 --branch feature/x" in c for c in cmds)
assert any("--recurse-submodules" in c for c in cmds)
assert any("checkout deadbeef" in c for c in cmds)
# extra-env exported before build
assert any("export FOO='bar baz'" in c and "export TOKEN=secret" in c for c in cmds)
assert any("cd /work && make all" in c for c in cmds)
# artifact stage uses the requested source dir
assert any("/work/build" in c for c in cmds)
def test_build_run_linux_host_source_dir(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
_patch(monkeypatch)
src = tmp_path / "src"
src.mkdir()
(src / "hello.txt").write_text("hi", encoding="utf-8")
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.5",
"--guest-os",
"linux",
"--host-source-dir",
str(src),
"--build-command",
"true",
"--ssh-key-path",
str(tmp_path / "key"),
],
)
assert result.exit_code == 0, result.output
instance = _FakeTransport.instances[0]
# A single tarball was uploaded.
assert len(instance.copies) == 1
local, remote = instance.copies[0]
assert local.endswith(".tar.gz")
assert remote.startswith("/tmp/")
# Extraction command was invoked.
assert any("tar -xzf" in c[0] for c in instance.runs)
def test_build_run_linux_skip_artifact(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
_patch(monkeypatch)
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.5",
"--guest-os",
"linux",
"--clone-url",
"https://example/repo.git",
"--build-command",
"make",
"--skip-artifact",
"--ssh-key-path",
str(tmp_path / "key"),
],
)
assert result.exit_code == 0, result.output
cmds = [c[0] for c in _FakeTransport.instances[0].runs]
# No staging command should run.
assert not any("/opt/ci/output" in c for c in cmds)
def test_build_run_linux_build_failure_exits_nonzero(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
_patch(monkeypatch)
# Make the build command (4th run: rm/git clone/checkout-skipped/build)
# return non-zero. Index of build run varies; easiest: make any non-zero
# result for the run whose script contains 'cd '.
src = tmp_path / "src"
src.mkdir()
class _BuildFails(_FakeTransport):
def run(self, script: str, *, check: bool = True) -> _FakeResult:
if "cd /opt/ci/build && make" in script:
self.runs.append((script, check))
return _FakeResult(stdout="boom\n", stderr="err\n", returncode=2)
return super().run(script, check=check)
monkeypatch.setattr(build_module, "SshTransport", _BuildFails)
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.5",
"--guest-os",
"linux",
"--host-source-dir",
str(src),
"--ssh-key-path",
str(tmp_path / "key"),
],
)
assert result.exit_code != 0
assert "build command failed (exit 2)" in result.output
# ── happy path: Windows ────────────────────────────────────────────────────
def test_build_run_windows_clone_url_default_dotnet(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
_patch(monkeypatch)
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.7",
"--guest-os",
"windows",
"--clone-url",
"https://example/repo.git",
"--use-shared-cache",
],
)
assert result.exit_code == 0, result.output
cmds = [c[0] for c in _FakeTransport.instances[0].runs]
assert any("git clone --depth 1" in c for c in cmds)
assert any("dotnet restore" in c for c in cmds)
assert any("NUGET_PACKAGES" in c for c in cmds)
def test_build_run_windows_host_source_dir(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
_patch(monkeypatch)
src = tmp_path / "src"
src.mkdir()
(src / "a.cs").write_text("//", encoding="utf-8")
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.7",
"--host-source-dir",
str(src),
"--build-command",
"echo built > dist\\out.txt",
"--guest-artifact-source",
"dist",
],
)
assert result.exit_code == 0, result.output
instance = _FakeTransport.instances[0]
assert len(instance.copies) == 1
assert instance.copies[0][1] == "C:\\CI\\src-transfer.zip"
cmds = [c[0] for c in instance.runs]
assert any("Expand-Archive" in c for c in cmds)
# Packaging step references the configured artifact source dir.
assert any("Compress-Archive" in c and "'dist'" in c for c in cmds)
+212
View File
@@ -0,0 +1,212 @@
"""Tests for ``ci_orchestrator vm new``.
Migrated from Pester ``tests/New-BuildVM.Tests.ps1``. Preserves:
* parameter validation: missing template VMX rejected
* failure path: backend clone failure cleans up the partial clone dir
* clone naming: ``Clone_{JobId}_{yyyyMMdd_HHmmss}``
* output: stdout final line is the clone VMX path
"""
from __future__ import annotations
from pathlib import Path
from typing import Any
import pytest
from click.testing import CliRunner
import ci_orchestrator.commands.vm as vm_module
from ci_orchestrator.__main__ import cli
from ci_orchestrator.backends.errors import BackendOperationFailed
from ci_orchestrator.backends.protocol import VmHandle
class _FakeBackend:
"""Records calls and creates the destination VMX on success."""
def __init__(self, *, fail: bool = False) -> None:
self.fail = fail
self.calls: list[dict[str, Any]] = []
def clone_linked(
self,
template: str,
snapshot: str,
name: str,
destination: str | None = None,
) -> VmHandle:
self.calls.append(
{
"template": template,
"snapshot": snapshot,
"name": name,
"destination": destination,
}
)
if self.fail:
raise BackendOperationFailed("clone", 1, "Error: source not found")
# Real vmrun creates dest dir + VMX file. Mirror that here so the
# post-clone existence check passes.
assert destination is not None
dst = Path(destination)
dst.parent.mkdir(parents=True, exist_ok=True)
dst.write_text("config.version = \"8\"", encoding="utf-8")
return VmHandle(identifier=str(dst), name=name)
@pytest.fixture
def fake_template(tmp_path: Path) -> Path:
p = tmp_path / "WinBuild2025.vmx"
p.write_text("config.version = \"8\"", encoding="utf-8")
return p
def test_vm_new_rejects_missing_template(tmp_path: Path) -> None:
"""Pester migration: throws when TemplatePath does not exist."""
missing = tmp_path / "ghost.vmx"
base = tmp_path / "build-vms"
result = CliRunner().invoke(
cli,
[
"vm",
"new",
"--template",
str(missing),
"--clone-base-dir",
str(base),
"--job-id",
"test-1",
],
)
assert result.exit_code != 0
assert "Template VMX not found" in result.output
def test_vm_new_clone_failure_cleans_up_partial_dir(
monkeypatch: pytest.MonkeyPatch, fake_template: Path, tmp_path: Path
) -> None:
"""Pester migration: vmrun failure removes the partially-created clone dir."""
base = tmp_path / "build-vms"
backend = _FakeBackend(fail=True)
monkeypatch.setattr(vm_module, "_make_backend", lambda _v: backend)
result = CliRunner().invoke(
cli,
[
"vm",
"new",
"--template",
str(fake_template),
"--clone-base-dir",
str(base),
"--job-id",
"test-fail",
],
)
assert result.exit_code != 0
assert "clone failed" in result.output
leftovers = [p for p in base.iterdir() if p.is_dir() and "test-fail" in p.name]
assert leftovers == []
def test_vm_new_clone_name_format_and_stdout(
monkeypatch: pytest.MonkeyPatch, fake_template: Path, tmp_path: Path
) -> None:
"""Pester migration: clone folder is ``Clone_{JobId}_{yyyyMMdd_HHmmss}``
and the final stdout line is the clone VMX path."""
base = tmp_path / "build-vms"
backend = _FakeBackend()
monkeypatch.setattr(vm_module, "_make_backend", lambda _v: backend)
result = CliRunner().invoke(
cli,
[
"vm",
"new",
"--template",
str(fake_template),
"--clone-base-dir",
str(base),
"--job-id",
"run-42",
],
)
assert result.exit_code == 0, result.output
final = result.output.strip().splitlines()[-1]
assert final.endswith(".vmx")
assert "Clone_run-42_" in final
# Format check: timestamp is 8 digits + underscore + 6 digits.
import re
assert re.search(r"Clone_run-42_\d{8}_\d{6}\.vmx$", final)
def test_vm_new_passes_snapshot_default(
monkeypatch: pytest.MonkeyPatch, fake_template: Path, tmp_path: Path
) -> None:
base = tmp_path / "build-vms"
backend = _FakeBackend()
monkeypatch.setattr(vm_module, "_make_backend", lambda _v: backend)
result = CliRunner().invoke(
cli,
[
"vm",
"new",
"--template",
str(fake_template),
"--clone-base-dir",
str(base),
"--job-id",
"run-99",
],
)
assert result.exit_code == 0, result.output
assert backend.calls[0]["snapshot"] == "BaseClean"
def test_vm_new_pascal_case_aliases(
monkeypatch: pytest.MonkeyPatch, fake_template: Path, tmp_path: Path
) -> None:
"""The PS-shim translation produces ``--template-path`` / ``--snapshot-name``."""
base = tmp_path / "build-vms"
backend = _FakeBackend()
monkeypatch.setattr(vm_module, "_make_backend", lambda _v: backend)
result = CliRunner().invoke(
cli,
[
"vm",
"new",
"--template-path",
str(fake_template),
"--snapshot-name",
"Custom",
"--clone-base-dir",
str(base),
"--job-id",
"alias",
],
)
assert result.exit_code == 0, result.output
assert backend.calls[0]["snapshot"] == "Custom"
def test_vm_new_creates_base_dir(
monkeypatch: pytest.MonkeyPatch, fake_template: Path, tmp_path: Path
) -> None:
base = tmp_path / "missing-base"
backend = _FakeBackend()
monkeypatch.setattr(vm_module, "_make_backend", lambda _v: backend)
result = CliRunner().invoke(
cli,
[
"vm",
"new",
"--template",
str(fake_template),
"--clone-base-dir",
str(base),
"--job-id",
"bd",
],
)
assert result.exit_code == 0, result.output
assert base.is_dir()