Simone
816a15503e
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.
290 lines
9.2 KiB
Python
290 lines
9.2 KiB
Python
"""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
|